Java实现生产者消费者

方法1：最简单--利用LinkedBlockingQueue

队列具有先进先出的特点，成为经常应用于生产-消费者模式的数据结构。
1.将一个对象放到队列尾部，如果队列已满，就等待直到有空闲节点。 ——put()方法
2.从队列头部取一个对象，如果没有对象，就等待直到有对象可取。 ——take()方法
3.在存取队列的过程中，锁定队列对象，不允许其它线程访问队列。——使得它是线程安全的

下面的代码适用于多个生产者、多个消费者。

//Producer.java
import java.util.concurrent.BlockingQueue;

/*
 *  Usage example, based on a typical producer-consumer scenario.
 * Note that a <tt>BlockingQueue</tt> can safely be used with multiple
 * producers and multiple consumers.
 * <pre>
 */
class Producer implements Runnable {
    private BlockingQueue<Object> queue;

    Producer(BlockingQueue<Object> q) {
        queue = q;
    }

    public void run() {
        try {
            while (true) {
                queue.put(produce());
                Thread.sleep(1);
                System.out.println("+1 Produce an Object:" + "生产者size:"
                        + queue.size());
            }
        } catch (InterruptedException ex) {
            System.out.println("生产者结束!");
        }
    }

    Object produce() {
        return new Object();
    }
}

//Consumer.java
import java.util.concurrent.BlockingQueue;

class Consumer implements Runnable {
    private BlockingQueue<Object> queue;

    Consumer(BlockingQueue<Object> q) {
        queue = q;
    }

    public void run() {
        try {
            while (true) {
                consume(queue.take());
                Thread.sleep(10);
                System.out.println("-1 Consumer an Object:" + "size:"+ queue.size());
            }
        } catch (InterruptedException ex) {
            System.out.println("消费者结束！");
        }
    }

    void consume(Object x) {
    }
}

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

class Main {
    public static void main(String[] args) {
        BlockingQueue<Object> q = new LinkedBlockingQueue<Object>();// SomeQueueImplementation();
        Producer p = new Producer(q);
        Consumer c1 = new Consumer(q);
        Consumer c2 = new Consumer(q);
        Thread pth = new Thread(p);
        pth.start();
        new Thread(c1).start();
        new Thread(c2).start();
        try {
            Thread.sleep(50);
            pth.interrupt();
        } catch (InterruptedException e) {
        }
    }
}

方法2：不采用BlockingQueue，利用lock、Condition来实现

思路：

利用可重用的LinkedList来存放资源，生产者addLast()，消费者removeFirst()。

使用ReentrantLock()来实现消费者和生产者的同步;//可重用的互斥锁

思考：为什么使用了lock后，还使用了它的2个newCondition，这样做有什么好处？

import java.util.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ProducerConsumer {
    private static final LinkedList<Integer> buffer = new LinkedList<Integer>();
    private static final int BUFFERSIZE = 10;
    private static Lock lock = new ReentrantLock();//可重用的互斥锁
    private static Condition NotFull = lock.newCondition();
    private static Condition NotEmpty = lock.newCondition();

    static class Producer extends Thread {
        public void run() {
            while (true) {
                // lock
                lock.lock();
                if (buffer.size() == BUFFERSIZE) {
                    System.out.println("BUffer is fulls");
                    try {
                        NotFull.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    buffer.addLast(1);
                    NotEmpty.signal();
                }
                lock.unlock();
                // unlock
            }
        }
    }
    
    static class Consumer extends Thread {
        public void run() {
            while (true) {
                // lock
                lock.lock();
                if (buffer.size() == 0) {
                    System.out.println("BUffer is empty");
                    try {
                        NotEmpty.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    buffer.removeFirst();
                    NotFull.signal();
                }
                // unlock
                lock.unlock();
            }
        }
    }
    
    public static void main(String[] args){
        new Consumer().start();
        new Producer().start();
    }
}

关于Java线程的lock、condition、synchronized的小小总结：

1.synchronized简单易用，但是功能受局限，无法满足复杂的同步机制，比如“读者写者问题”：多个读者线程是不互斥的。

2.lock以更优雅的方式来解决线程同步，比如读写锁ReadWriteLock。

3.Condition解决线程间通信，为不同的线程建立不同的条件。

推荐博文：Java线程-锁机制

Condition可以替代传统的线程间通信，用await()替换wait()，用signal()替换notify()，用signalAll()替换notifyAll()。

     ——为什么方法名不直接叫wait()/notify()/nofityAll()？因为Object的这几个方法是final的，不可重写！

Condition实现了BlockingQueue的功能。

看看BlockingQueue的继承和实现。

一般的锁Lock的实现：注意ReadLock、WriteLock是内部的静态类，只有ReentrantReadWriteLock类可以调用。

读写的锁ReadWriteLock

思考：OS的信号量是什么，线程通信的方式有哪些？共享变量和消息传递。

        信号量与互斥锁有什么联系？

        参考资料 

By BYRHuangQiang 2014-03-18 09:05:30