Java并发编程实践 目录
并发编程—— 闭锁CountDownLatch 与 栅栏CyclicBarrier
并发编程—— CompletionService : Executor 和 BlockingQueue
概述
第1 部分 问题描述
第2 部分 日志服务实例
参考
第1 部分 问题描述
应用程序通常会创建拥有多个线程的服务，例如线程池，并且这些服务的生命周期通常比创建它们的方法的生命周期更长。如果应用程序准备退出，那么这些服务所拥有的线程也需要结束。由于无法通过抢占式的方法来停止线程，因此它们需要自行结束。
正确的封装原则是：除非拥有某个线程，否则不能对该线程进行操控。例如，中断线程或者修改线程的优先级等。
与其他的封装对象一样，线程的所有权是不可传递的：应用程序可以拥有服务，服务也可以拥有工作者线程，但应用程序并不能拥有工作者线程，因此应用程序不能停止工作者线程。相反，服务应该提供生命周期方法，服务就可以关闭所有的线程了。这样，当应用程序关闭服务时，服务就可以关闭所有的线程了。
对于持有线程的服务，只要服务的存在时间大于创建线程的方法的存在时间，那么就应该提供生命周期方法。
第2 部分 日志服务实例
下面程序给出一个日志服务示例，其中日志操作在单独的日志线程中执行。产生日志消息的线程并不会将消息直接写入输出流，而是 LogWriter 通过 BlockingQueue 将消息提交给日志线程，并由日志线程写入。这是一种多生产者单消费者的设计方式：每个调用 log 的操作都相当于一个生产者，而后台的日志线程则相当于消费者。
1 /**
2 * 不支持关闭的生产者-消费者日志服务
3 */
4 public class LogWriter {
5 private final BlockingQueue<String> queue;
6 private final LoggerThread logger;
7
8 public LogWriter(Writer writer){
9 this.queue = new LinkedBlockingDeque<String>();
10 this.logger = new LoggerThread(writer);
11 }
12
13 public void start(){
14 logger.start();
15 }
16
17 public void log(String msg) throws InterruptedException{
18 queue.put(msg);
19 }
20
21 private class LoggerThread extends Thread{
22 private final Writer writer;
23
24 public LoggerThread(Writer writer) {
25 this.writer = writer;
26 }
27
28 @Override
29 public void run() {
30 try {
31 while(true){
32 writer.write(queue.take());
33 }
34 } catch (IOException e) {
35 // io exception handle
36 } catch (InterruptedException e) {
37 // interrupt exceptino handle
38 } finally{
39 try {
40 writer.close();
41 } catch (IOException e) {
42 e.printStackTrace();
43 }
44 }
45 }
46 }
47 }
下面给出 向LogWriter 添加可靠的取消操作的程序：
1 /**
2 * 7.15 向LogWriter 添加可靠的取消操作
3 *
4 * @ClassName: LogService
5 * @author xingle
6 * @date 2014-10-24 下午4:35:57
7 */
8 public class LogService {
9
10 private final BlockingQueue<String> queue;
11 private final LoggerThread loggerThread;
12 private final PrintWriter writer;
13 @GuardedBy("this")
14 private boolean isShutdown;
15 @GuardedBy("this")
16 private int reservations;
17
18 public LogService(Writer writer) {
19 this.queue = new LinkedBlockingDeque<String>();
20 this.loggerThread = new LoggerThread();
21 this.writer = new PrintWriter(writer);
22 }
23
24 public void start() {
25 loggerThread.start();
26 }
27
28 public void stop() {
29 synchronized (this) {
30 isShutdown = true;
31 }
32 loggerThread.interrupt();
33 }
34
35 /**
36 * 为LogService 提供可靠关闭操作的方法是解决竞态条件问题，因而要使日志消息的提交操作作为原子操作，
37 * 然而 ，不希望在消息加入队列时去持有一个锁，因为 put 方法本身就可以阻塞
38 * 这里 采用的方法是：通过原子方式来检查关闭请求，并且有条件地递增一个计数器来“保持”提交消息的权利
39 */
40 public void log(String msg) throws InterruptedException {
41 synchronized (this) {
42 if (isShutdown)
43 throw new IllegalStateException();
44 ++reservations;
45 }
46 queue.put(msg);
47 }
48
49 /**
50 * 消费日志线程
51 */
52 private class LoggerThread extends Thread {
53 public void run() {
54 try {
55 while (true) {
56 try {
57 synchronized (LogService.this) {
58 if (isShutdown && reservations == 0)
59 break;
60 }
61 String msg = queue.take();
62 synchronized (LogService.this) {
63 --reservations;
64 }
65 writer.println(msg);
66 } catch (InterruptedException e) {
67 /* retry */
68 }
69
70 }
71 } finally {
72 writer.close();
73 }
74 }
75 }
76 }
参考
1.《并发编程》 7.2 停止基于线程的服务
时间： 2024-08-10 21:28:03