并发编程—— 任务取消 之 停止基于线程的服务

并发编程—— ConcurrentHashMap

并发编程—— 阻塞队列和生产者-消费者模式

并发编程—— 闭锁CountDownLatch 与 栅栏CyclicBarrier

并发编程—— Callable和Future

并发编程—— CompletionService : Executor 和 BlockingQueue

并发编程—— 任务取消

并发编程—— 任务取消 之 中断

并发编程—— 任务取消 之 停止基于线程的服务

 

概述

第1 部分 问题描述

第2 部分 日志服务实例

参考 

第1 部分 问题描述

　　应用程序通常会创建拥有多个线程的服务，例如线程池，并且这些服务的生命周期通常比创建它们的方法的生命周期更长。如果应用程序准备退出，那么这些服务所拥有的线程也需要结束。由于无法通过抢占式的方法来停止线程，因此它们需要自行结束。

正确的封装原则是：除非拥有某个线程，否则不能对该线程进行操控。例如，中断线程或者修改线程的优先级等。

　　与其他的封装对象一样，线程的所有权是不可传递的：应用程序可以拥有服务，服务也可以拥有工作者线程，但应用程序并不能拥有工作者线程，因此应用程序不能停止工作者线程。相反，服务应该提供生命周期方法，服务就可以关闭所有的线程了。这样，当应用程序关闭服务时，服务就可以关闭所有的线程了。

对于持有线程的服务，只要服务的存在时间大于创建线程的方法的存在时间，那么就应该提供生命周期方法。

 

第2 部分 日志服务实例

　　下面程序给出一个日志服务示例，其中日志操作在单独的日志线程中执行。产生日志消息的线程并不会将消息直接写入输出流，而是 LogWriter 通过 BlockingQueue 将消息提交给日志线程，并由日志线程写入。这是一种多生产者单消费者的设计方式：每个调用 log 的操作都相当于一个生产者，而后台的日志线程则相当于消费者。

/**
 * 不支持关闭的生产者-消费者日志服务
 */
public class LogWriter {
    private final BlockingQueue<String> queue;
    private final LoggerThread logger;
     
    public LogWriter(Writer writer){
        this.queue = new LinkedBlockingDeque<String>();
        this.logger = new LoggerThread(writer);
    }
     
    public void start(){
        logger.start();
    }
     
    public void log(String msg) throws InterruptedException{
        queue.put(msg);
    }
 
    private class LoggerThread extends Thread{
        private final Writer writer;
         
        public LoggerThread(Writer writer) {
            this.writer = writer;
        }
 
        @Override
        public void run() {
            try {
                while(true){
                    writer.write(queue.take());
                }
            } catch (IOException e) {
                // io exception handle
            } catch (InterruptedException e) {
                // interrupt exceptino handle
            } finally{
                try {
                    writer.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

 

下面给出 向LogWriter 添加可靠的取消操作的程序：

/**
 * 7.15 向LogWriter 添加可靠的取消操作
 * 
 * @ClassName: LogService 
 * @author xingle
 * @date 2014-10-24 下午4:35:57
 */
public class LogService {

    private final BlockingQueue<String> queue;
    private final LoggerThread loggerThread;
    private final PrintWriter writer;
    @GuardedBy("this")
    private boolean isShutdown;
    @GuardedBy("this")
    private int reservations;

    public LogService(Writer writer) {
        this.queue = new LinkedBlockingDeque<String>();
        this.loggerThread = new LoggerThread();
        this.writer = new PrintWriter(writer);
    }

    public void start() {
        loggerThread.start();
    }

    public void stop() {
        synchronized (this) {
            isShutdown = true;
        }
        loggerThread.interrupt();
    }

    /**
     * 为LogService 提供可靠关闭操作的方法是解决竞态条件问题，因而要使日志消息的提交操作作为原子操作，
     * 然而 ，不希望在消息加入队列时去持有一个锁，因为 put 方法本身就可以阻塞
     * 这里 采用的方法是：通过原子方式来检查关闭请求，并且有条件地递增一个计数器来“保持”提交消息的权利
     */
    public void log(String msg) throws InterruptedException {
        synchronized (this) {
            if (isShutdown)
                throw new IllegalStateException();
            ++reservations;
        }
        queue.put(msg);
    }

    /**
     * 消费日志线程
     */
    private class LoggerThread extends Thread {
        public void run() {
            try {
                while (true) {
                    try {
                        synchronized (LogService.this) {
                            if (isShutdown && reservations == 0)
                                break;
                        }
                        String msg = queue.take();
                        synchronized (LogService.this) {
                            --reservations;
                        }
                        writer.println(msg);
                    } catch (InterruptedException e) {
                        /* retry */
                    }

                }
            } finally {
                writer.close();
            }
        }
    }
}

 

---

参考 

1.《并发编程》 7.2 停止基于线程的服务