线程基础（三十三）

本文作者：王一飞，叩丁狼高级讲师。原创文章，转载请注明出处。

接上篇,本篇讲解线程另外一个设计模式:Producer-Consumer Pattern.

概念

Producer是生产,生产者的意思, 指生产数据的线程, Consumer 则是消费,消费者的意思. 指使用数据的线程.

Producer-Consumer 模式主要目标:生产者生产数据能安全得交给消费者进行处理.

当Producer 跟 Consumer 都为一个线程时, Producer-Consumer 模式就变为Pipe模式

参与角色

Producer-Consumer模式参与角色:
Data: 资源对象(数据对象)
Data角色由Producer角色生成,供Consumer角色使用,一般都需要放置到一个容器(Channel角色)中进行安全管理.

Producer: 生产者
Producer角色生成Data角色,并将其传递给Channel角色存放, Consumer角色再从Channel角色中获取Data角色

Consumer:消费者
Consumer角色从Channel角色中获取Producer角色生成Data角色, 并使用

Channel:通道
Channel角色保管Producer角色生成的Data角色,还能响应Consumer角色处理Data角色的请求,传递Data角色,一般而言,为了确保Data的安全性, Channel角色对Producer角色跟Consumer角色的访问执行互斥处理.

当Channel角色满足接收Data角色条件时, 才接纳Producer角色生成的Data角色, 否则Producer角色一直等待,直到条件满足.
当Channel角色满足传递Data角色条件时,才将Data角色传递给Consumer角色,否则Consumer角色一直等待,直到条件满足.

image.png

模式特征:

1:存在数据创建对象(producter)
2:存在数据使用对象(consumer)
3:存在数据存储对象(channel)
4:存在被操作的对象(data)
5:存在操作条件控制(channel接受与传递data条件)

演示案例

需求:5个生产者生产Data数据, 5个消费者消费Data数据, data至多只能存放10个, 超过10个, 生产者等待, 少于1个消费者等待.

//数据
public class Data {
    public void show(){
        System.out.println("data对象 dosomething....");
    }
}

//通道
public class Channel {
    private int count;  //channel存储data个数
    @Getter
    private List<Data> list = new ArrayList<Data>();

    public Channel(int count){
        this.count = count;
    }
    public synchronized void setData(Data data) throws InterruptedException {
        if(list.size() < count){
            list.add(data);
            System.out.println(Thread.currentThread().getName() + " 生产了一个数据, 目前数据个数:" +list.size());
            notifyAll();  //唤醒所有等待的消费者
        }else{
            System.out.println(Thread.currentThread().getName() + " 检查目前数据个数:" +list.size() +" 不需要生产,等待....");
            wait(); //超过容量,等待
        }
    }

    public synchronized Data getData() throws InterruptedException {
        Data data = null;
        while(true){ //某个消费者被唤醒后,可以直接进行消费,而不是空返回
            if (list.size() == 0){
                System.out.println(Thread.currentThread().getName() + " 检查目前数据个数:" +list.size() +" 不能消费,等待....");
                wait(); //等待生产者生产数据
            }else{
                data = list.remove(0);//从最底拿起;
                System.out.println(Thread.currentThread().getName() + " 消费了一个数据, 目前数据个数:"+list.size());
                notifyAll(); //唤醒所有等待的消费者
                break;
            }
        }
        return data;
    }

}

//生产者
public class Producter implements Runnable {

    private Channel channel;
    public Producter(Channel channel){
        this.channel = channel;
    }

    public void run() {
        while (true){
            try {
                channel.setData(new Data());
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

//消费者
public class Consumer implements Runnable {

    private Channel channel;
    public Consumer(Channel channel){
        this.channel = channel;
    }


    public void run() {
       while (true){
           try {
               channel.getData().show();
               Thread.sleep(100);
           } catch (InterruptedException e) {
               e.printStackTrace();
           }
       }
    }
}

public class App {

    public static void main(String[] args) {
        Channel channel = new Channel(10);
        for (int i = 0; i < 5; i++) {
            new Thread(new Producter(channel), "producter_" + i).start();
        }
        for (int i = 0; i < 5; i++) {
            new Thread(new Consumer(channel), "consumer_" + i).start();
        }

    }
}

producter_0 生产了一个数据, 目前数据个数:1
producter_3 生产了一个数据, 目前数据个数:2
producter_2 生产了一个数据, 目前数据个数:3
producter_4 生产了一个数据, 目前数据个数:4
producter_1 生产了一个数据, 目前数据个数:5
consumer_0 消费了一个数据, 目前数据个数:4
data对象 dosomething....
consumer_1 消费了一个数据, 目前数据个数:3
data对象 dosomething....
consumer_2 消费了一个数据, 目前数据个数:2
data对象 dosomething....
consumer_3 消费了一个数据, 目前数据个数:1
data对象 dosomething....
consumer_4 消费了一个数据, 目前数据个数:0
data对象 dosomething....
producter_2 生产了一个数据, 目前数据个数:1
producter_0 生产了一个数据, 目前数据个数:2
producter_3 生产了一个数据, 目前数据个数:3
producter_4 生产了一个数据, 目前数据个数:4
consumer_3 消费了一个数据, 目前数据个数:3
data对象 dosomething....
consumer_2 消费了一个数据, 目前数据个数:2
data对象 dosomething....
consumer_1 消费了一个数据, 目前数据个数:1
data对象 dosomething....
consumer_0 消费了一个数据, 目前数据个数:0
data对象 dosomething....

几个注意点

1:为什么要存在Channel角色
是否可以将Producter生产的Data数据直接传递给Consumer消费, 跳过Channel这个角色?
想法是好的, 操作性不强, 少了channel中间传递角色, producter生产数据后, 势必需要调用consumer的方法, 执行消费动作, 那么producter生产跟消费便是同一个线程, 这个线程, 先执行生产,然后再执行消费逻辑. 那么这就失去异步操作优势了.

2:Channel 角色的安全性
chanel角色类中的getData与setData方法都使用synchronized 修饰,保证list 数据的操作安全.

3:Channel角色中获取数据的顺序问题
案例中, 使用的ArrayList集合存放Data数据,getData时,从最底位置list.remove(0)获取, 实际运用中可以结合需求,采用不同存储结构:
队列:先进先出
栈:后进先出
优先队列: 优先级高的先出.

4:如果 Producter 跟 Consumer 只有1个时,会怎样?
当Producter跟Consumer都为一个时, 便是管道模式(Pipe模式)

producter_0 生产了一个数据, 目前数据个数:1
consumer_0 消费了一个数据, 目前数据个数:0
data对象 dosomething....
consumer_0 检查目前数据个数:0 不能消费,等待....
producter_0 生产了一个数据, 目前数据个数:1
consumer_0 消费了一个数据, 目前数据个数:0
data对象 dosomething....

一生产,马上使用.经典运用案例:
SynchronousQueue类的实现逻辑

适用场景

Producer-Consumer Pattern 是最实用的异步操作模式, 其操作思想广泛运用于各类MQ中间件中. 而jdk中很多相关集合类中也使用生产者-消费者模式, 比如:
实现BlockingQueue接口的阻塞队列:
ArrayBlockingQueue: 基于数字的阻塞队列
LinkedBlockingQueue: 基于链表的阻塞队列
ProrityBlockingQueue:基于优先级的阻塞队列
DelayQueue: 具有延时特效的阻塞队列
SynchronousQueue:直接传递的阻塞队列
ConcurrentLinkedQueue:没有个数限制的阻塞队列