聊聊生产者消费者模型

今天聊聊算不上设计模式的设计模式——生产者与消费者模型。

为什么说算不上设计模式的设计模式呢？

因为对我来说，现在用的最熟练的就是这个了。其他23中设计模式大概读了读，但是代码实际使用中基本没用到过，仅仅根据个人当时脑袋一热就打出来代码了。

废话不多说，今天就来主要介绍下最常用的多线程同步模型——生产者与消费者。

介绍

生产者与消费者模型，顾名思义需要有两个角色——生产者和消费者。

最简单的形式就是有一个生产者，一个消费者。

生产者产生数据后直接给消费者进行消费，但是如果生产者和消费者两个速度不一致的话就会有一方等待。

那么怎么来减少等待的时间呢，于是我们就引进来了缓冲队列。生产者与消费者不直接进行“通信”，两者有个中间人来传达数据，这样就完美的解决了两方速度不一致的时候会有一方等待的问题。

有了这个缓冲队列，生产者只需关系自己的生产而不用关心消费者的消费能力；消费者也只管消费，不用知道生产者如何生产数据的更不用等待生产者生产完再消费。

缓冲队列

目前java中有现成的缓冲队列，而在本文中则主要介绍两种常用的——ArrayBlockingQueue和LinkBlockingQueue。

blockingqueue，听声辨义可知是阻塞队列，其最大的功能就是能够阻塞的插入与删除队列元素同时在队列空的时候能够阻塞消费者线程，队列满的时候能够阻塞生产者线程。

ArrayBlockingQueue由循环数组实现的，目的是为了减少数组空间频繁的分配与释放。初始化的时候就需要指定数组分配空间大小。

LinkBlockingQueue由列表实现，无界，初始化的时候不需要指定空间。

常用的操作方式：

1
2
3
4
5
6

add  如果队列满了则抛出异常，java.lang.IllegalStateException: Queue full
pffer  如果队列满了，程序继续执行，队列中不会增加数据
put 如果队列满了会阻塞，直到队列中有空间再执行put
poll 弹出队顶元素，队列为null的时候返回空
peek 返回队顶元素，但不弹出
take  弹出队顶元素，为null的时候阻塞消费者

常用的方式已经了解了，现在来聊聊生产者消费者模型中经常遇到的问题吧。

问题

1. 大部分情况下消费者基本都处于while(true)的判断条件下，即无限循环等待新的任务到来，那么怎么优雅的在所有任务都结束后kill掉消费者呢？

目前最流行的方式就是在生产者全部生产完成后在分别给每一个缓冲队列发送结束标志，告知消费者消费完这个就该停止了。

代码实现：

1
2
3
4
5
6
7
8
9
10
11
12
13

CountDownLatch countDownLatch = new CountDownLatch(5);
try {
    countDownLatch.await();
} catch (InterruptedException e) {
    log.error("主线程等待被中断",e);
}
try {
    for (Integer key : arrayBlockingQueueMap.keySet()){
        arrayBlockingQueueMap.get(key).put(NewRead.Stop);
    }
} catch (InterruptedException e) {
    log.error("停止被中断",e);
}

每一个生产者都在执行完毕后把计时器减一，当计时器归0的时候主线程就会往所有的缓冲队列中放入停止的标志。

1
2
3

finally {
            countDownLatch.countDown();
        }

消费者判断数据类型是不是停止的标志，如果是立马退出，否则继续消费。

1
2
3
4
5
6
7

Object object = arrayBlockingQueue.take();
if (object == NewRead.POISON_PILL){
log.info("分片" + shardingValue + "导数完成");
break;
}else{
///继续消费
}

应该还有其他友好的实现方式，评论区可以讨论，我是不知道了(¬､¬)。