Java并发编程实战之基于生产者消费者模式的日志服务读书笔记

日常开发中，我们会经常跟日志系统打交道，日志系统算是典型的生产者消费者模型的系统。（多个不同的）生产者线程负责生产日志消息，这些系统扮演者生产者的角色，日志系统负责将消息队列里的日志上传到服务器，属于消费者角色。是一种多生产者单消费者模式的设计方式。如果生产者的生产速度大于LoggerThread 的消费速度，则BlockingDeque线程会阻塞生产者，直到LoggerThread 有能力处理新的消息。

不支持关闭的生产者-消费者日志服务（bug版本）

public class LogService {private final BlockingQueue<String> queue;private final LoggerThread loggerThread;public LogService() {//注意队列的容量为10queue = new LinkedBlockingQueue<>(10);loggerThread = new LoggerThread();}public void start() {loggerThread.start();}/*** 生产者生产消息，每个log的调用者都行相当于一个生产者** @param msg* @throws InterruptedException*/public void log(String msg) throws InterruptedException {queue.put(msg);}/*** 消费者线程消费消息，该线程扮演的是消费者这一角色。*/private class LoggerThread extends Thread {public  LoggerThread(){}public void run() {try {while (true) {//无限循环System.out.println(queue.take());}//end while} catch (InterruptedException e) {//在循环外响应中断} finally {System.out.println("log service is close");}}}}

上面代码看起来很正常，但是有如下缺点：
1、LoggerThread没有提供关闭的方法，当没有生产者生产消息的时候，那么LoggerThread在调用queue.take()的时候会一直处于阻塞状态，导致JVM没法正常关闭。比如下面的调用代码，就会发生LoggerThread无法关闭的情况。

 public static void main(String[] args) {LogService logService = new LogService();for(int i=0;i<100;i++) {final int index =i;new Thread(new Runnable() {@Overridepublic void run() {try {logService.log("生产者生产消息===="+index);} catch (InterruptedException e) {System.out.println("response interception");}}}).start();}logService.start();}

关闭LoggerThread很容易啊，只需要把LoggerThread的方法改成如下就可以了吧？

 public void run() {try {while (!queue.isEmpty()) {//判断队列是否为空，如果为空则关闭System.out.println(queue.take());}} catch (InterruptedException e) {System.out.println("response interception");} finally {System.out.println("log service is close="+queue.size());}}

上面的改动中，我们将while(true)改成了while(!queue.isEmpty()).可以吗？答案是完全不可以！在我们这个demo中，队列的大小是有限制的，如果消费者的消费速度大于生产者的速度，那么队列在某一瞬间就变成空了，也就是queue.isEmpty()为true，此时消费者关闭，但是生产者可以继续生产，则队列满的情况下生产者会阻塞。取消一个生产者-消费者模型的最可靠的方法是需要同时取消生产者和消费者，因为当单独的关闭消费者的时候，会阻塞生产者；当单独的关闭生产者的时候，会阻塞消费者。其关闭原则应该是：我们应该提供一个Stop方法，调用stop方法的时候表示日志服务已经关闭，此时生产者不可以再生产消息，当消费者消费完队列里的消息后才正式关闭日志服务，避免造成日志的丢失，

且看下面代码，我们为LogService提供了stop方法，并且修改了log方法，当isShutDown为true的时候，就不能在生产消息。然而这看似完美的操作却有个巨大漏洞，stop方法不是原子的，if(!isShutdown)也不是原子的，使得关闭方法并不可靠

  private boolean isShutdown=false;public void stop(){isShutdown=true;}public void log(String msg) throws InterruptedException {if(!isShutdown){queue.put(msg);}}

所以我们需要改造。使之成为原子的即可。我们改造如下

支持关闭的生产者-消费者日志服务（不安全版本）

使用synchronized关键字，将stop改造成原子的，需要注意的是，因为put方法本身就可以阻塞，所以我们不需要在消息加入队列的时候再去持有一个锁，所以我们将put方法放在了synchronized语句块的外面。

private boolean isShutDown = false;public void stop(){synchronized(this) { isShutDown=true;}}/*** 生产者生产消息，每个log的调用者都行相当于一个生产者** @param msg* @throws InterruptedException*/public void log(String msg) throws InterruptedException {synchronized(this) {if(isShutDown) {return;}}queue.put(msg);       }

经过这么以改进，当我们调用stop的时候，则生产者线程不会继续生产消息。我们可以将LoggerThread改造如下：

 public void run() {try {while (true) {if(isShutDown) {//如果关闭则退出break;}System.out.println(queue.take());}} catch (InterruptedException e) {//注意我们这里实在while循环外响应中断的。System.out.println("response interception");} finally {System.out.println("log service is close="+queue.size());}}

那么这样就没问题了吗？答案是问题大了去了。比如我们调用了stop方法，此时LoggerThread消费者正好在queue.take()阻塞着，那么if(isShutDown) {break}则永远不会执行，LoggerThread是无法退出的。正确的做法是我们需要再次对stop方法进行改造，改造如下：

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

我们添加了loggerThread.interrupt();，此时take方法发生阻塞时会响应中断，因为我们在while循环之外响应中断的，则跳出while循环，执行打印如下：

  public static void main(String[] args) {LogService logService = new LogService();for(int i=0;i<90;i++) {final int index =i;Thread thread=new Thread(new Runnable() {@Overridepublic void run() {             try {               logService.log("生产者生产消息===="+index); } catch (InterruptedException e) {System.out.println("response interception");}               }       });thread.start();}logService.start();try {Thread.sleep(2000);} catch (InterruptedException e) {        }//两秒之后关闭日志系统logService.stop();}

执行打印如下：

response interception
log service is close=0

那么到此为止是不是就没问题了呢？当然仍然有问题，如果发生中断的时候，我们在while循环之外响应中断，消息队列中仍然有消息，我们就没法继续处理剩下的日志，那么日志服务的关闭势必造成日志的丢失，如果是比较关键的日志信息丢失了，我们都没地方哭去。

所以我们的日志服务系统仍然有优化的空间。

支持关闭的生产者-消费者日志服务（安全版本）

我们的目标是生产者生产多少消息，我们的日志系统就处理多少消息。当日志服务关闭的时候，生产者不能生产消息。消费者在处理完所有消息后，自行关闭。
为此我们继续改造，使用一个变量logCount，每调用一次log则logCount+1;注意logCount递增操作需要是原子的。
为此我们对log方法进行改造，改造如下：

    private int logCount=0;public void log(String msg) throws InterruptedException {synchronized(this) {if(isShutDown) {return;}//消息递增logCount++;}queue.put(msg);}

处理完了成产者，我们需要对消费者进行改造，注意LoggerThread之前的消费者捕获InterruptedException 时，是在while循环之外，我们现在需要放在循环之内。这样我们才能正确处理中断，确保队列里的消息处理完毕，改造后的LoggerThread如下：

到此为止，本篇博文就结束了，在读《Java并发编程实战》这一章节的时候，还有点疑问为什么这么写。直到写完了这篇博客，边写博客边敲程序进行验证，才算是彻底了解了其中的意义。果然写博客还是很有帮助的。起码这边博客对博主自己深入理解Java并发编程，有很大帮助。

完整的代码如下：

package log;import java.io.PrintWriter;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;public class LogService {private final BlockingQueue<String> queue;private final LoggerThread loggerThread;public LogService() {queue = new LinkedBlockingQueue<>(10);loggerThread = new LoggerThread();}public void start() {loggerThread.start();}private boolean isShutDown = false;public void stop() {synchronized (this) {isShutDown = true;}loggerThread.interrupt();}private int logCount = 0;public void log(String msg) throws InterruptedException {synchronized (this) {if (isShutDown) {return;}logCount++;}queue.put(msg);}/*** 消费者线程消费消息，该线程扮演的是消费者这一角色。*/private class LoggerThread extends Thread {public LoggerThread() {}public void run() {try {while (true) {try {synchronized (LogService.this) {if (isShutDown && logCount == 0) {break;}}System.out.println(queue.take());synchronized (LogService.this) {logCount--;}} catch (InterruptedException e) {System.out.println("相应中断，while循环继续执行，除非满足退出条件");}}//end while} finally {//资源额释放System.out.println("log service is close=" + queue.size());}}}public static void main(String[] args) {LogService logService = new LogService();for (int i = 0; i < 90; i++) {final int index = i;Thread thread = new Thread(new Runnable() {@Overridepublic void run() {try {logService.log("生产者生产消息====" + index);} catch (InterruptedException e) {System.out.println("response interception");}}});thread.start();}logService.start();try {Thread.sleep(2000);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();}logService.stop();}
}