Java 并发包下的提供Lock,Lock相对于Synchronized可以更好的解决线程同步问题,更加的灵活和高效,并且ReadWriteLock锁还能实现读、写的分离。但线程间仅仅互斥是不够的,还需要通信,本篇的内容是基于上篇之上,使用Lock如何处理线程通信。阻塞队列(BlockingQueue)就是使用condition的和lock实现的。可以查看:Java并发编程-阻塞队列(BlockingQueue)的实现原理
Condition
那么引入本篇的主角,Condition,Condition 将 Object的通信方法(wait、notify 和 notifyAll)分解成截然不同的对象,以便通过将这些对象与任意 Lock 实现组合使用,为每个对象提供多个等待 set (wait-set)。其中,Lock 替代了 synchronized 方法和语句的使用,Condition 替代了 Object 通信方法的使用。
在Condition中,用await()替换wait(),用signal()替换notify(),用signalAll()替换notifyAll(),传统线程的通信方式,Condition都可以实现,这里注意,Condition是被绑定到Lock上的,要创建一个Lock的Condition必须用newCondition()方法。
Condition的强大之处在于它可以为多个线程间建立不同的Condition, 使用synchronized/wait()只有一个阻塞队列,notifyAll会唤起所有阻塞队列下的线程,而使用lock/condition,可以实现多个阻塞队列,signalAll只会唤起某个阻塞队列下的阻塞线程。
下面用两种方式编写生产者/消费者模式代码加以说明。
- 使用synchronized/wait()实现生产者消费者模式如下:
//模拟生产和消费的对象
class Buffer {
private int maxSize;
private List<Date> storage;
Buffer(int size){
maxSize=size;
storage=new LinkedList<>();
}
//生产方法
public synchronized void put() {
try {
while (storage.size() ==maxSize ){//如果队列满了
System.out.print(Thread.currentThread().getName()+": wait \n");;
wait();//阻塞线程
}
storage.add(new Date());
System.out.print(Thread.currentThread().getName()+": put:"+storage.size()+ "\n");
Thread.sleep(1000);
notifyAll();//唤起线程
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
//消费方法
public synchronized void take() {
try {
while (storage.size() ==0 ){//如果队列满了
System.out.print(Thread.currentThread().getName()+": wait \n");;
wait();//阻塞线程
}
Date d=((LinkedList<Date>)storage).poll();
System.out.print(Thread.currentThread().getName()+": take:"+storage.size()+ "\n");
Thread.sleep(1000);
notifyAll();//唤起线程
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
//生产者
class Producer implements Runnable{
private Buffer buffer;
Producer(Buffer b){
buffer=b;
}
@Override
public void run() {
while(true){
buffer.put();
}
}
}
//消费者
class Consumer implements Runnable{
private Buffer buffer;
Consumer(Buffer b){
buffer=b;
}
@Override
public void run() {
while(true){
buffer.take();
}
}
}
//
public class Main{
public static void main(String[] arg){
Buffer buffer=new Buffer(10);
Producer producer=new Producer(buffer);
Consumer consumer=new Consumer(buffer);
//创建线程执行生产和消费
for(int i=0;i<3;i++){
new Thread(producer,"producer-"+i).start();
}
for(int i=0;i<3;i++){
new Thread(consumer,"consumer-"+i).start();
}
}
}
- 使用lock/condition实现生产者消费者模式如下:
import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
class Buffer {
private final Lock lock;
private final Condition notFull;
private final Condition notEmpty;
private int maxSize;
private List<Date> storage;
Buffer(int size){
//使用锁lock,并且创建两个condition,相当于两个阻塞队列
lock=new ReentrantLock();
notFull=lock.newCondition();
notEmpty=lock.newCondition();
maxSize=size;
storage=new LinkedList<>();
}
public void put() {
lock.lock();
try {
while (storage.size() ==maxSize ){//如果队列满了
System.out.print(Thread.currentThread().getName()+": wait \n");;
notFull.await();//阻塞生产线程
}
storage.add(new Date());
System.out.print(Thread.currentThread().getName()+": put:"+storage.size()+ "\n");
Thread.sleep(1000);
notEmpty.signalAll();//唤醒消费线程
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}finally{
lock.unlock();
}
}
public void take() {
lock.lock();
try {
while (storage.size() ==0 ){//如果队列满了
System.out.print(Thread.currentThread().getName()+": wait \n");;
notEmpty.await();//阻塞消费线程
}
Date d=((LinkedList<Date>)storage).poll();
System.out.print(Thread.currentThread().getName()+": take:"+storage.size()+ "\n");
Thread.sleep(1000);
notFull.signalAll();//唤醒生产线程
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}finally{
lock.unlock();
}
}
}
class Producer implements Runnable{
private Buffer buffer;
Producer(Buffer b){
buffer=b;
}
@Override
public void run() {
while(true){
buffer.put();
}
}
}
class Consumer implements Runnable{
private Buffer buffer;
Consumer(Buffer b){
buffer=b;
}
@Override
public void run() {
while(true){
buffer.take();
}
}
}
public class Main{
public static void main(String[] arg){
Buffer buffer=new Buffer(10);
Producer producer=new Producer(buffer);
Consumer consumer=new Consumer(buffer);
for(int i=0;i<3;i++){
new Thread(producer,"producer-"+i).start();
}
for(int i=0;i<3;i++){
new Thread(consumer,"consumer-"+i).start();
}
}
}
- 当生产者执行put方法时,调用notEmpty.signalAll()只会唤醒notEmpty.await()下的消费者线程。
- 当消费者执行塔克方法时,调用notFull.signalAll()只会唤醒notFull.await()下的消费者线程。
更新
- 2019.5.4
- object锁,可以通过使用多个不同对象作为锁,实现多个阻塞队列。