Java多线程实现生产者消费者程序（Wait，Notify实现和Lock，Condition实现）

Java多线程实现生产者消费者程序。

如下代码是反例，只适用于一个生产者和消费者的情景。对于多个生产者消费者的情景如下实现代码是错误的！！！（错误原因是对线程加锁了，而没有对临界区加锁，记录下来并不是想误导别人，就想提醒自己曾经犯过的错误）。正确的生产者消费者见下面部分！！！ （如果存在错误欢迎指导提出错误）



使用wait和notify的一个生产者一个消费者的实现：

package com.daxin;

import java.util.ArrayList;
import java.util.Random;
/**
*
* 生产者消费者程序
*
* @author Daxin
*
*/
public class ProducerAndConsumer {

private ArrayList<String> values = new ArrayList<String>();//共享数据临界区

public static void main(String[] args) throws InterruptedException {

ProducerAndConsumer pc = new ProducerAndConsumer();

Producer p = pc.new Producer();

Consumer c = pc.new Consumer();

p.start();
c.start();

}

class Producer extends Thread {

@Override
public void run() {

while (true) {// 一直生产

synchronized (values) {//进行加锁互斥访问
if (values.size() == 10) {
System.out.println("队列满了,无法生产数据");
try {
values.wait();//阻塞生产者生产，和消费者的notify对应，消费者的notify唤醒这一行的wait
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
values.notify();
}

}
//需要在同步代码块中实现
String str = "" + new Random().nextInt(20);
System.out.println("生产了数据为 " + str);
values.add(str);
values.notify();// 唤醒消费者去交费

}
}

}

}

class Consumer extends Thread {
@Override
public void run() {

while (true) {// 一直消费

synchronized (values) {//进行加锁互斥访问
if (values.size() == 0) {
//
System.out.println("没有数据了,无法完成消费了");
try {

values.wait();// 消费者进行等待
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
values.notify();
}
}
//需要在同步代码块中实现
String value = values.remove(0);
System.out.println("消费了数据为：" + value);
values.notify();// 通知生产者生产消息

}
}

}
}

}

使用并发重入锁ReentrantLock和Condition的一个生产者一个消费者的实现：

package com.daxin;

import java.util.ArrayList;
import java.util.Random;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class ProducerAndConsumerReentrantLockCondition {
ReentrantLock rl = new ReentrantLock();
Condition cd = rl.newCondition();
private ArrayList<String> values = new ArrayList<String>();// 共享数据临界区

public static void main(String[] args) {
ProducerAndConsumerReentrantLockCondition pc = new ProducerAndConsumerReentrantLockCondition();
ProducerReentrantLockCondition p = pc.new ProducerReentrantLockCondition();
ConsumerReentrantLockCondition c = pc.new ConsumerReentrantLockCondition();

p.start();
c.start();

}

class ProducerReentrantLockCondition extends Thread {

@Override
public void run() {
while (true) {
rl.lock();// 生产者获取锁进行互斥访问
if (values.size() == 10) {
try {
System.out.println("生产满了");
cd.await();// 阻塞生产者进行生产
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
rl.unlock();
}

}

String e = (new Random().nextInt(20)) + "";
System.out.println("生产者生产数据为 " + e);
values.add(e);

cd.signal();// 唤醒消费者生产

}

}

}

class ConsumerReentrantLockCondition extends Thread {
@Override
public void run() {
while (true) {
rl.lock();

if (values.size() == 0) {

try {
System.out.println("消费空了");
cd.await();// 阻塞消费者进行消费
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}

}

String e = values.remove(0);

System.out.println("消费的数据为 " + e);

cd.signal();// 唤醒消费者消费

rl.unlock();

}

}
}

}

上面代码是错误的代码示例，作为提醒避免犯这种错误。

上面代码不健全的原因就是对线程加锁了。所以如果在多个生产者和多个消费者的模型中，程序的声明周期中只有第一个获取到所的生产者和第一个获取锁的消费者能够相互等待唤醒，其他线程在程序声明周期内是一直等待的！因此解决这个问题就是创建临界区，然后对临界区进行加锁：

健全的wait和notify的实现

/**
* 缓冲区容量为10
*
* 使用notify和wait实现的缓冲区
*
*/
class Buffer_WN {

private List<Integer> buffer = new ArrayList<>();

private final int capacity = 10;
private  int sum = 0;

/**
* synchronized对当前this实例加锁。静态方法对class字节码文件加锁。
*
* @throws Exception
*/
public void put(String name) throws Exception {
// 对临界区进行加锁
synchronized (buffer) {

// while解决伪唤醒
while (buffer.size() == capacity) {
buffer.wait();
}
if((++sum)>20)
{
System.out.println("数据生产完毕！");
return ;
}
buffer.add(8);
System.out.println(name + " 生产了一条数据 +.");
buffer.notifyAll();
}

}

public  void get(String name) throws Exception {
// 对临界区进行加锁
synchronized (buffer) {

// while解决伪唤醒
while (buffer.size() == 0) {
buffer.wait();
}

buffer.remove(0);
System.out.println(name + " 消费了一条数据 -.");
buffer.notifyAll();
}

}
}

/**
*
*
* @author Daxin
*
*/
public class ProducerAndConsumer_Wait_Notify {

// 锁：是互斥变量 。控制互斥访问时候资源不可用时候进行阻塞
static Lock lock = new ReentrantLock();
// condition：条件变量，操作系统的。当不满足需要的条件时候进行互斥。
static Condition c = lock.newCondition();

static List<Integer> buffer = new ArrayList<>();

public static void main(String[] args) {
Buffer_WN buffer = new Buffer_WN();
Thread producer = newProducer("1111", buffer);
producer.start();
Thread producer1 = newProducer("2222", buffer);
producer1.start();

Thread consumer = newConsumer("3333", buffer);
consumer.start();
Thread consumer1 = newConsumer("4444", buffer);
consumer1.start();

}

public static Thread newProducer(final String name, final Buffer_WN buffer) {
return new Thread(new Runnable() {

@Override
public void run() {

try {
while (true) {

buffer.put(name);
Thread.sleep(500);
}
} catch (Exception e) {
e.printStackTrace();
}

}

});
}

public static Thread newConsumer(final String name, final Buffer_WN buffer) {
return new Thread(new Runnable() {

@Override
public void run() {

try {
while (true) {

buffer.get(name);
Thread.sleep(500);
}
} catch (Exception e) {

e.printStackTrace();
}

}
});

}

}

健全的ReentrantLock和Condition实现方式1，对Lock绑定一个Condition对象（对临界区加锁），并使用if 判断解决伪唤醒：

package com.daxin.lock.condition;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
* 缓冲区容量为10
* @author Daxin
*
*/
class Buffer {

private List<Integer> buffer = new ArrayList<>();
// 锁：是互斥变量 。控制互斥访问时候资源不可用时候进行阻塞
Lock lock = new ReentrantLock();
// condition：条件变量，操作系统的。当不满足需要的条件时候进行互斥。
Condition c = lock.newCondition();

public void put(String name) {

try {
lock.lock();

if (buffer.size() == 10) {
//代码位置1
c.await(); // 缓冲区满了的话则阻塞当前线程在此处，等待消费者线程通知唤醒
}

//避免生产者1，在代码位置1处await，然后生产者2在代码位置2处恰好生产一个元素导致buffer是10个。所以执行代码位置3，而代码位置3恰好唤醒了生产者1，因此导致buffer缓冲区大于10
if (buffer.size() < 10) {// 也就是所谓解决伪唤醒，也可以使用while循环解决伪唤醒
int val = new Random().nextInt(100);
buffer.add(new Integer(val));//代码位置2
System.out.println("生产者"+name+" 生产数据完毕！生产的数据为：" + val + "缓冲区大小为：" + buffer.size());
}

//			代码位置3
c.signalAll();// 如果有资源可以获得的话，通知消费者消费（注意：此时如果当前缓冲区没有满的话，不会调用当前的c.await();
// 所以不会释放锁，所以消费者线程也无法消费，只有缓冲区满了之后才可以获取锁进行消费）
Thread.sleep(200);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} finally {
lock.unlock();
}

}

public void get(String name) {
try {

lock.lock();

if (buffer.size() == 0) {
c.await(); // 阻塞，等待生产者生产

}

if (buffer.size() > 0) { //解决伪唤醒代码
int val = buffer.get(0);
System.out.println("消费者"+name+" 消费数据完毕！消费数据为：" + val + "缓冲区大小为：" + buffer.size());
buffer.remove(0);

}

c.signalAll();
Thread.sleep(200);
}

catch (

Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}

}
}

/**
*
*
* @author Daxin
*
*/
public class ProducerAndConsumer {

// 锁：是互斥变量 。控制互斥访问时候资源不可用时候进行阻塞
static Lock lock = new ReentrantLock();
// condition：条件变量，操作系统的。当不满足需要的条件时候进行互斥。
static Condition c = lock.newCondition();

static List<Integer> buffer = new ArrayList<>();

public static void main(String[] args) {
Buffer buffer = new Buffer();
Thread producer = newProducer("1", buffer);
producer.start();
Thread producer1 = newProducer("2", buffer);
producer1.start();

Thread consumer = newConsumer("1", buffer);
consumer.start();
Thread consumer1 = newConsumer("2", buffer);
consumer1.start();

}

public static Thread newProducer(final String name, final Buffer buffer) {
return new Thread(new Runnable() {

@Override
public void run() {

while (true) {

buffer.put(name);

}

}

});
}

public static Thread newConsumer(final String name, final Buffer buffer) {
return new Thread(new Runnable() {

@Override
public void run() {

while (true) {

buffer.get(name);

}

}
});

}

}

健全的ReentrantLock和Condition实现方式1，对Lock绑定一个Condition对象（对临界区加锁），并使用while解决伪唤醒：

package com.daxin.lock.condition;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
*
*
* @author Daxin
*
*/
class Buffer_WHX {

private List<Integer> buffer = new ArrayList<>();
// 锁：是互斥变量 。控制互斥访问时候资源不可用时候进行阻塞
Lock lock = new ReentrantLock();
// condition：条件变量，操作系统的。当不满足需要的条件时候进行互斥。
Condition c = lock.newCondition();// 最好还是参见JDK文档使用两个Condition实例

public void put(String name) {

try {
lock.lock();

while (buffer.size() == 10) { //while死循环解决伪唤醒
// 代码位置1
c.await(); // 缓冲区满了的话则阻塞当前线程在此处，等待消费者线程通知唤醒
}

int val = new Random().nextInt(100);
buffer.add(new Integer(val + 20));// 代码位置2
System.out.println("生产者" + name + " 生产数据完毕！生产的数据为：" + val + "缓冲区大小为：" + buffer.size());

// 代码位置3
c.signalAll();// 如果有资源可以获得的话，通知消费者消费（注意：此时如果当前缓冲区没有满的话，不会调用当前的c.await();
// 所以不会释放锁，所以消费者线程也无法消费，只有缓冲区满了之后才可以获取锁进行消费）

} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}

}

public void get(String name) {
try {

lock.lock();

while (buffer.size() == 0) {
c.await(); // 阻塞，等待生产者生产

}

int val = buffer.get(0);
System.out.println("消费者" + name + " 消费数据完毕！消费数据为：" + val + "缓冲区大小为：" + buffer.size());
buffer.remove(0);

c.signalAll();

}

catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}

}
}

/**
*
*
* @author Daxin
*
*/
public class ProducerAndConsumer_4 {

// 锁：是互斥变量 。控制互斥访问时候资源不可用时候进行阻塞
static Lock lock = new ReentrantLock();
// condition：条件变量，操作系统的。当不满足需要的条件时候进行互斥。
static Condition c = lock.newCondition();

static List<Integer> buffer = new ArrayList<>();

public static void main(String[] args) {
Buffer buffer = new Buffer();
Thread producer = newProducer("1111", buffer);
producer.start();
Thread producer1 = newProducer("2222", buffer);
producer1.start();

Thread consumer = newConsumer("3333", buffer);
consumer.start();
Thread consumer1 = newConsumer("4444", buffer);
consumer1.start();

}

public static Thread newProducer(final String name, final Buffer buffer) {
return new Thread(new Runnable() {

@Override
public void run() {

while (true) {

buffer.put(name);

}

}

});
}

public static Thread newConsumer(final String name, final Buffer buffer) {
return new Thread(new Runnable() {

@Override
public void run() {

try {
while (true) {

buffer.get(name);
}
} catch (Exception e) {

e.printStackTrace();
}

}
});

}

}

健全的ReentrantLock和Condition实现方式2，对Lock绑定两个Condition对象（对临界区加锁）：

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

/**
* JDK文档携带的Condition示例
*
* @author Daxin
*
*/
class BoundedBuffer {
final Lock lock = new ReentrantLock();
// Condition 将 Object 监视器方法（wait、notify 和 notifyAll）分解成截然不同的对象
final Condition notFull = lock.newCondition();
// Condition 将 Object 监视器方法（wait、notify 和 notifyAll）分解成截然不同的对象
final Condition notEmpty = lock.newCondition();

final Object[] items = new Object[5];
int putptr, takeptr, count;

public void put(Object x) throws InterruptedException {
lock.lock();
try {
while (count == items.length)
// 满了的话阻塞此处，并释放锁。
notFull.await();
items[putptr] = x;
if (++putptr == items.length)
putptr = 0;
++count;
notEmpty.signal();
} finally {
// 没有满的话。执行finally释放锁
lock.unlock();
}
}

public Object take() throws InterruptedException {
lock.lock();
try {
while (count == 0)
notEmpty.await();
Object x = items[takeptr];
if (++takeptr == items.length)
takeptr = 0;
--count;
notFull.signal();
return x;
} finally {
lock.unlock();
}
}
}

public class JDK_Document_Lock_Condition {

public static void main(String[] args) throws Exception {
final BoundedBuffer bf = new BoundedBuffer();
new Thread(new Runnable() {

@Override
public void run() {
try {
while(true){
bf.put(new Object());
System.out.println("生产者生产完毕！");
}
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}

}
}).start();

new Thread(new Runnable() {

@Override
public void run() {
try {

while(true){
System.out.println("消费者消费完毕！ "+bf.take());
}
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}

}
}).start();

}
}

Lock和Condition在JDK中LinkedBlockingQueue的应用，核心源码如下：

import java.util.concurrent.LinkedBlockingQueue.Node;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
* LinkedBlockingQueue核心方法源码分析
*
*
*/
public class LinkedBlockingQueue {

/** Current number of elements */
//使用AtomicInteger的原因是：LinkedBlockingQueue的take和put使用的是两把锁。所以需要对count进行同步。
//同时count使用AtomicInteger可以解决take和put的冲突操作
private final AtomicInteger count = new AtomicInteger(0);

/** Lock held by take, poll, etc */
private final ReentrantLock takeLock = new ReentrantLock();

/** Wait queue for waiting takes */
private final Condition notEmpty = takeLock.newCondition(); //绑定takeLock

/** Lock held by put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock();

/** Wait queue for waiting puts */
private final Condition notFull = putLock.newCondition();//绑定putLock

/**
* Inserts the specified element at the tail of this queue, waiting if
* necessary for space to become available.
*
* @throws InterruptedException
*             {@inheritDoc}
* @throws NullPointerException
*             {@inheritDoc}
*/
public void put(E e) throws InterruptedException {
if (e == null)
throw new NullPointerException();
// Note: convention in all put/take/etc is to preset local var
// holding count negative to indicate failure unless set.
int c = -1;
Node<E> node = new Node(e);
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
/*
* Note that count is used in wait guard even though it is not
* protected by lock. This works because count can only decrease at
* this point (all other puts are shut out by lock), and we (or some
* other waiting put) are signalled if it ever changes from
* capacity. Similarly for all other uses of count in other wait
* guards.
*/
while (count.get() == capacity) {
notFull.await(); //绑定putLock
}
enqueue(node);
// 此处 c =size - 1（size为容器实际大小）
c = count.getAndIncrement();
if (c + 1 < capacity) //c+1 =size-1+1 =size ，如果c + 1 =size < capacity 的话
notFull.signal(); //唤醒其他生产者生产数据
} finally {
putLock.unlock();
}
if (c == 0) //c=size-1==0，就是size==1？如果size=1代表还有元素，通知消费者生产数据
signalNotEmpty(); //notEmpty.signal();
}

/**
* Signals a waiting take. Called only from put/offer (which do not
* otherwise ordinarily lock takeLock.)
*/
private void signalNotEmpty() {
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
notEmpty.signal(); //绑定takeLock
} finally {
takeLock.unlock();
}
}

public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
while (count.get() == 0) {
notEmpty.await();  //绑定takeLock
}
x = dequeue();
// 有点绕：获取当前count的值，然后减1。
c = count.getAndDecrement();
//而此时由于x = dequeue()代码消费了一个数据，所以c=size+1
if (c > 1)//当前c=size+1>1的话，则就是size>0.所以此时就是有元素的，唤醒其他消费者消费数据
notEmpty.signal();  //绑定takeLock
} finally {
takeLock.unlock();
}
if (c == capacity) //c = size +1 ==capacity 所以size =capacity-1，则容器还没有满，所以通知生产者生产数据
signalNotFull();  // 调用notFull.signal();
return x;
}

/**
* Signals a waiting put. Called only from take/poll.
*/
private void signalNotFull() {
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
notFull.signal(); //绑定putLock
} finally {
putLock.unlock();
}
}
}