《Java并发编程实战》使用newTaskFor实现线程取消的疑惑

Question

问题描述

在《Java并发编程实战》第7章，作者通过newTaskFor方法封装实现了线程的取消。给出的demo如下：

public abstract class SocketUsingTask <T> implements CancellableTask<T> {
    @GuardedBy("this") private Socket socket;

    protected synchronized void setSocket(Socket s) {
        socket = s;
    }

    public synchronized void cancel() {
        try {
            if (socket != null)
                socket.close();
        } catch (IOException ignored) {
        }
    }

    public RunnableFuture<T> newTask() {
        return new FutureTask<T>(this) {
            public boolean cancel(boolean mayInterruptIfRunning) {
                try {
                    SocketUsingTask.this.cancel();
                } finally {
                    return super.cancel(mayInterruptIfRunning);
                }
            }
        };
    }
}


interface CancellableTask <T> extends Callable<T> {
    void cancel();

    RunnableFuture<T> newTask();
}


@ThreadSafe
class CancellingExecutor extends ThreadPoolExecutor {
    public CancellingExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    public CancellingExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
    }

    public CancellingExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
    }

    public CancellingExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }

    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        if (callable instanceof CancellableTask)
            return ((CancellableTask<T>) callable).newTask();
        else
            return super.newTaskFor(callable);
    }
}

我觉得这段代码有点绕，感觉很不直白，我个人感觉可以换一种方法实现：

public abstract class SocketUsingTask<T> extends FutureTask<T> implements Callable<T> {
    @GuardedBy("this") private Socket socket;
    
    protected synchronized void setSocket(Socket s) { socket = s; }
    
    public synchronized void cancel() {
        try {
            if (socket != null)
                socket.close();
        } catch (IOException ignored) { }
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        try {
            cancel();
        } finally {
            return super.cancel(mayInterruptIfRunning);
        }
    }
}

我想知道《Java并发编程实战》中给出的这个demo的最大的优点在哪？为什么要这样实现？因为感觉存在一些多余的操作。谢谢

Answer 1

你应该看一下AbstractExecutorService的submit以及整个继承链的代码

    /**
     * @throws RejectedExecutionException {@inheritDoc}
     * @throws NullPointerException       {@inheritDoc}
     */
    public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task);
        execute(ftask);
        return ftask;
    }

上面的代码submit相当于调用CancellingExecutor的newTaskFor因为你子类重写了。书中的demo只需要写一个类继承SocketUsingTasks重写call即可，或匿名实现也行。然后放入CancellingExecutor执行。

Answer 2

我觉得demo的写法有2点：
第一demo把Callable<T>重新用 CancellableTask<T>扩展了一下，如果你要增加新的功能就可以做到面向接口编程的效果，你直接实现了扩展性肯定没这个好；
第二我看newTaskFor那个就是生成 RunnableFuture<T>的工厂，这个demo中貌似没有用到

Answer 3

像这样的类似问题可以提很多，比如ArrayList为什么要继承List接口，直接继承Collection接口也能直接实现
软件工程有一些很重要的实践原则，比如开闭原则，决定了你的代码的扩展性如何，一开始不理解是很正常的，都是实践的经验。项目又不是能完成当时的需求即可，还需要考虑的后期的扩展和维护的