首页 > 代码库 > Java 多线程学习笔记:生产者消费者问题

Java 多线程学习笔记:生产者消费者问题

前言:最近在学习Java多线程,看到ImportNew网上有网友翻译的一篇文章《阻塞队列实现生产者消费者模式》。在文中,使用的是Java的concurrent包中的阻塞队列来实现。在看完后,自行实现阻塞队列。

(一)准备

  在多线程中,生产者-消费者问题是一个经典的多线程同步问题。简单来说就是有两种线程(在这里也可以做进程理解)——生产者和消费者,他们共享一个固定大小的缓存区(如一个队列)。生产者负责产生放入新数据,消费者负责取出缓存区的数据。具体介绍请参考 Producer-consumer problem 。

 

(二)设计概述

  详细的代码可以到我的 Github 仓库下载(master分支为详细的代码,输出所有的必要信息;simplification分支为简化的代码)。

  项目代码,分为四个类:Consumer类为消费者线程,Producer类为生产者线程,ProducerConsumer测试的启动类,ProducerConsumerQueue类为阻塞队列。

  在多线程中,生产者不断向队列放入新数据,当队列已满时,生产者将被阻塞,直到消费者从队列中取出部分数据并唤醒生产者;消费者不断从队列取出数据,当队列为空时,消费者将被阻塞,直到生产者再次放入新数据并唤醒消费者。在JAVA的concurrent包中,提供了一系列阻塞队列帮我们完成这一工作。这一工作也恰好是生产者消费者的问题的关键所在。在这里,我将自行实现阻塞队列:ProducerConsumerQueue;

(三)详细实现

  (1)ProducerConsumerQueue类

 1 public final class ProducerConsumerQueue<E> { 2     //声明一个重入锁 3     private final Lock lock = new ReentrantLock(); 4     private final Condition notEmpty = lock.newCondition(); 5     private final Condition notFull = lock.newCondition(); 6     //缓存大小 7     private final int CAPACITY; 8     //存储数据的缓存队列 9     private Queue<E> queue;10 11     public ProducerConsumerQueue(int CAPACITY) {12         this.CAPACITY = CAPACITY;13         this.queue = new LinkedList<E>();14     }15 16     //put the e into the queue17     public void put(E e) throws InterruptedException {18         lock.lock();19 20         try {21             while (queue.size() == this.CAPACITY)22                 this.notFull.await();23             queue.offer(e);24             this.notEmpty.signalAll();25         } finally {26             lock.unlock();27         }28     }29 30     //get e from queue.31     public E take() throws InterruptedException {32         lock.lock();33 34         try {35             while (queue.size() == 0)36                 this.notEmpty.await();37             this.notFull.signalAll();38             return queue.poll();39         } finally {40             lock.unlock();41         }42     }43 }

 

 

  该队列提供put()和take()方法,前者用于生产者放入数据,后者用于消费者取出数据。

  在put方法中,首先要获得锁。然后检测是否队列已满,如果是,阻塞生产者线程,并释放锁;当再次获得锁时,这里使用了while语句,使得线程再次检查队列是否已满,如果未满,放入数据到缓存,并唤醒可能阻塞的消费者线程。在这里使用while,目的是防止其他生产者线程已经唤醒,并且成功放入新数据到队列,然后阻塞,而该线程又再次放入,导致可能缓存溢出。

  在take方法中,同样要先获得锁。检测队列是否已空,如果是,阻塞消费者线程,释放锁;当被唤醒并再次获得锁时,使用while再次检查队列是否仍为空,如果非空,则取出缓存数据。while同样为了防止其他消费者对一个空队列取数据。

  (2)Consumer类

 1 public class Consumer implements Runnable { 2     private final ProducerConsumerQueue sharedQueue; 3  4     public Consumer(ProducerConsumerQueue sharedQueue) { 5         this.sharedQueue = sharedQueue; 6     } 7  8     @Override 9     public void run() {10         while (true) {11             try {12                 synchronized (this){13                     System.out.println("\tConsumer: " + sharedQueue.take());14                 }15             } catch (InterruptedException ex) {16                 System.out.println(Thread.currentThread().getName() + " throw a interrupexception.");17                 //do something.18             }19         }20     }21 }

 

  消费者线程持续从队列中取出数据。

  (3)Producer类

 1 public class Producer implements Runnable { 2     private final ProducerConsumerQueue sharedQueue; 3  4     public Producer(ProducerConsumerQueue sharedQueue) { 5         this.sharedQueue = sharedQueue; 6     } 7  8     @Override 9     public void run() {10         for (int i = 0; i < 20; i++) {11             try {12                 synchronized (this){13                     System.out.println("Producer: "+i);14                     sharedQueue.put(i);15                 }16             } catch (InterruptedException ex) {17                 System.out.println(Thread.currentThread().getName() + " throw a interrupexception. ");18                 //do something.19             }20         }21     }22 }

  生产者线程产生数据并放入队列中。

  (4)ProducerConsumer测试类

public class ProducerConsumer {    public static void main(String[] args) {        ProducerConsumerQueue sharedQueue = new ProducerConsumerQueue(4);        Thread prodThread = new Thread(new Producer(sharedQueue));        Thread consThread1 = new Thread(new Consumer(sharedQueue));        Thread consThread2 = new Thread(new Consumer(sharedQueue));        prodThread.start();        consThread1.start();        consThread2.start();    }}

 

    创建一个生产者线程和两个消费者线程。

 

扩展:

1、Wikipedia:http://en.wikipedia.org/wiki/Producer%E2%80%93consumer_problem

2、《聊聊并发——生产者和消费者模式》http://www.infoq.com/cn/articles/producers-and-consumers-mode

Java 多线程学习笔记:生产者消费者问题