Java程序设计：生产者和消费者模型

以经典的生产者和消费者模型为例来讲解多线程的并发协作。该模型简单描述为：一群生产者线程和一群消费者线程共享同一个初始为空、存放固定数据数量的缓冲区。生产者的任务是在缓冲区没满时，不断生产数据并放入缓存区，如果缓存区已满，则生产者必须等待。同时，只有当缓冲区不为空时，消费者才能从缓冲区中不断取出一个或多个数据（即将数据从缓冲区中移出），如果缓冲区为空，表明没有数据可消费，则消费者必须等待。可见，缓冲区是所有生产者和消费者的共享资源，且一次只允许一个线程访问。

缓冲区是实现并发协作的核心。有了缓冲区，可以更好地解耦生产者和消费者，生产者无须同消费者直接打交道，生产者只要看到缓冲区没满，就可以向缓冲区里放置数据。反之亦然，消费者只要看到缓冲区里不为空，就可以从缓冲区里取数据。由于生产者和消费者在被阻塞后，需要被其他线程唤醒，因此涉及线程间通信问题。

【例6-10】

演示生产者和消费者模型。

步骤1：定义生产和消费的数据类My Data。生产者和消费者生产的数据类My Data可以按照需要定义，这里只提供一个序列号属性，如图6-27所示。

图6-27　数据类MyData定义

步骤2：设计缓冲区DataBuffer类。缓冲区DataBuffer类的设计如图6-28所示。

图6-28　定义缓冲区DataBuffer类

（1）利用java.util.Array List＜E＞类来实现自定义缓冲区DataBuffer类。Array List＜E＞类是一个泛型类，是一个按照线性表结构实现的列表，内部是通过数组实现的，在项目9集合中详细讲解，这里简单理解下，Array List＜E＞类表示列表里面的每个元素的数据类型都是E占位符代表的类型，创建一个列表时E需要指定。

（2）为了设置缓冲区可容纳数据的数量，采用一个final修饰的int型变量capacity来表示，该值通过构造方法参数设定。构造方法根据capacity的值来开辟缓冲区空间，如果该参数不是正整数，则将抛出一个运行时异常Illegal Argument Exception。

（3）缓冲区DataBuffer类除了定义成员属性、构造方法外，还提供了获取缓冲区数据数量的get Quantity方法、判断缓冲区是否已满的isFull方法、判断缓冲区是否为空的isEmpty方法、生产一个数据（在列表尾部添加）的produce方法、消费一个数据（先进先出，先消费列表头部的数据）的consume方法。

步骤3：设计生产者任务Producer Task类。生产者任务Producer Task类如图6-29所示。

图6-29　定义生产者任务Producer Task类(www.xing528.com)

（1）在生产者任务类Producer Task中，定义DataBuffer型成员变量buffer，通过构造方法接收和保存缓冲区对象。

（2）生产者生产的所有数据产品都有一个顺序编号，通过static关键字修饰的int型变量id表示待生产数据的序列号，初始化为1。static关键字修饰意味着变量id被该类所有对象共享。

（3）所有生产者和消费者对共享资源缓冲区对象需要互斥访问，需要考虑线程同步问题。可以使用缓冲区对象buffer作为锁对象，也可以重新定义一个所有生产者线程和消费者线程都使用的同一个锁对象。

（4）当一个生产者线程在获得buffer锁对象后发现缓冲区是满的时候，生产者线程需要加入buffer锁对象的等待池中等待，随后该线程进入阻塞状态，等待buffer锁对象上的其他线程唤醒。当被唤醒时，由阻塞状态转为就绪状态，等待被调度运行。由于可能存在多个生产者线程，一旦线程被唤醒，需要再次判断缓冲区是否是满的，这里通过while循环来实现，如果是满的，重新进入阻塞状态。当缓冲区没有满时，生产者线程可以生产数据，在控制台打印相关信息，并将下一个待生产数据的序列号加1，然后通过锁对象buffer调用notify All方法唤醒该对象上所有的等待线程。生产者生产了数据，意味着缓冲区不再是空的，所有消费者线程都可以进行消费任务。当然，如果缓冲区没有满，所有生产者线程也可以继续生产。注意，无论是哪个线程向缓冲区生产数据或消费数据，都必须先获得同一个buffer锁对象。

步骤4：设计消费者任务Consumer Task类。消费者任务Consumer Task类如图6-30所示。

图6-30　定义消费者任务Consumer Task类

（1）在消费者任务类Consumer Task中，定义DataBuffer型变量buffer，通过构造方法接收和保存缓冲区对象。

（2）当一个消费者线程获取到锁对象buffer后发现缓冲区是空的时候，意味着消费者没有数据可消费，需要加入buffer对象等待池中等待，该线程随即处于阻塞状态，等待buffer对象上的其他线程唤醒。当被唤醒时，由阻塞状态转为就绪状态，等待被调度运行。由于可能存在多个生产者和消费者线程，一旦线程被唤醒，需要再次判断缓冲区是否是空的，这里通过while循环来实现，如果是空的，重新进入阻塞状态。当缓冲区不为空时，消费者线程可以消费数据，并在控制台打印相关信息，然后通过锁对象buffer调用notify All方法唤醒该对象上所有的等待线程。消费者消费了数据，意味着缓冲区不再是满的，生产者线程也可以进行生产任务。注意，无论是哪个线程向缓冲区生产数据或消费数据，都必须先获得同一个buffer锁对象。

步骤5：编写测试类Producer Comsumer Test。

下面通过一个测试用例来测试生产者和消费者模型，如图6-31所示。创建一个可容纳2个数据的缓冲区，然后创建一个生产者任务对象和消费者任务对象，接着创建2个生产者线程和2个消费者线程（模拟多个生产者、多个消费者场景）。

图6-31　生产者和消费者模型测试

结果显示，各线程之间实现了并发协作。由于生产者线程和消费者线程会一直继续下去，所以程序需要强制退出。