JAVA线程间协作:Condition

JAVA线程间协作:Condition

内置条件队列存在一些缺陷。每个内置锁都只能有一个相关联的条件队列,因而在像BounderBuffer这种类中,多个线程可能在同一个条件队列上等待不同的条件谓词,并且在最常见的加锁模式下公开条件队列对象。这些因素都使得无法满足在使用notifyAll时所有等待线程为统一类型的需求。如果想编写一个带有多个条件谓词的并发对象,或者想获得除了条件队列可见性之外的更多控制权,就可以使用显示的Lock和Condition而不是内置锁和条件队列,这是一种更灵活的选择。

一个Condition和一个Lock关联在一起,就想一个条件队列和一个内置锁相关联一样。要创建一个Condition,可以在相关联的Lock上调用Lock.newCondition方法。正如Lock比内置加锁提供了更为丰富的功能,Condition同样比内置条件队列提供了更丰富的功能:在每个锁上可存在多个等待、条件等待可以是可中断的或者不可中断的、基于时限的等待,以及公平的或非公平的队列操作。

与内置条件队列不同的是,对于每个Lock,可以有任意数量的Condition对象。Condition对象继承了相关的Lock对象的公平性,对于公平的锁,线程会依照FIFO顺序从Condition.await中释放。

Condition接口:

public interface Condition{
void await() throws InterruptedException;
boolean await(long time, TimeUnit unit) throws InterruptedException;
long awaitNanos(long nanosTimeout) throws InterruptedException;
void awaitUniterruptibly();
boolean awaitUntil(Date deadline) throws InterruptedException;
void signal();
void signalAll();
}

注意:在Condition对象中,与wait,notify和notifyAll方法对于的分别是await,signal,signalAll。但是,Condition对Object进行了扩展,因而它也包含wait和notify方法。一定要确保使用的版本——await和signal.
官方使用方法如下:

class BoundedBuffer
{
final Lock lock = new ReentrantLock();
final Condition notFull = lock.newCondition();
final Condition notEmpty = lock.newCondition();

final Object[] items = new Object[100];
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
{
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();
}
}
}

或者通过一个实际的例子来解释Condition的用法:
我们要打印1-9这0个数字,由A线程先打印1-3,然后由B线程打印4-6,然后再由A线程打印7-9.我们采用Condition来演示解决方法:

package com.cooperation;

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

public class TestCondition
{
private static int value = 1;
private Lock lock = new ReentrantLock();
private Condition Condition456 = lock.newCondition();
private Condition Condition789 = lock.newCondition();

class ThreadA implements Runnable{
@Override
public void run()
{
try
{
lock.lock();
System.out.println("首先输出1-3");
while(value<=3)
{
System.out.println(value++);
}
Condition456.signal();
}
finally
{
lock.unlock();
}

try
{
lock.lock();
while(value<=6)
{
Condition789.await();
}
System.out.println("输出7-9");
while(value<=9)
{
System.out.println(value++);
}
}
catch (InterruptedException e)
{
e.printStackTrace();
}
finally
{
lock.unlock();
}

}
}

class ThreadB implements Runnable{
@Override
public void run()
{
try
{
lock.lock();
while(value<=3)
{
Condition456.await();
}
}
catch (InterruptedException e)
{
e.printStackTrace();
}
finally{
lock.unlock();
}

try{
lock.lock();
System.out.println("输出4-6");
while(value<=6)
{
System.out.println(value++);
}
Condition789.signal();
}
finally
{
lock.unlock();
}
}
}

public static void main(String[] args)
{
TestCondition test = new TestCondition();
Thread threadA = new Thread(test.new ThreadA());
Thread threadB = new Thread(test.new ThreadB());
threadA.start();
threadB.start();
}

}

输出结果:

首先输出1-3
1
2
3
输出4-6
4
5
6
输出7-9
7
8
9

如果需要采用Object方法的wait,notify,notifyAll方法实现这个实例可以参考:http://outofmemory.cn/java/java.util.concurrent/thread-sync-with-object-wait-notify-notifyAll

同样可以扩张,ThreadA打印123,ThreadB打印456,再让ThreadA打印789,最后然ThreadB打印10 11 12.如下:

package com.cooperation;

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

public class TestCondition
{
private static int value = 1;
private Lock lock = new ReentrantLock();
private Condition Condition456 = lock.newCondition();
private Condition Condition789 = lock.newCondition();
private Condition Condition101112 = lock.newCondition();

class ThreadA implements Runnable{
@Override
public void run()
{
try
{
lock.lock();
System.out.println("首先输出1-3");
while(value<=3)
{
System.out.println(value++);
}
Condition456.signal();
}
finally
{
lock.unlock();
}

try
{
lock.lock();
while(value<=6)
{
Condition789.await();
}
System.out.println("输出7-9");
while(value<=9)
{
System.out.println(value++);
}
Condition101112.signal();
}
catch (InterruptedException e)
{
e.printStackTrace();
}
finally
{
lock.unlock();
}

}
}

class ThreadB implements Runnable{
@Override
public void run()
{
try
{
lock.lock();
while(value<=3)
{
Condition456.await();
}
}
catch (InterruptedException e)
{
e.printStackTrace();
}
finally{
lock.unlock();
}

try{
lock.lock();
System.out.println("输出4-6");
while(value<=6)
{
System.out.println(value++);
}
Condition789.signal();
}
finally
{
lock.unlock();
}

try
{
lock.lock();
while(value<=9)
{
Condition101112.await();
}
}
catch (InterruptedException e)
{
e.printStackTrace();
}
finally{
lock.unlock();
}

try{
lock.lock();
System.out.println("输出10-12");
while(value<=12)
{
System.out.println(value++);
}
}
finally
{
lock.unlock();
}
}
}

public static void main(String[] args)
{
TestCondition test = new TestCondition();
Thread threadA = new Thread(test.new ThreadA());
Thread threadB = new Thread(test.new ThreadB());
threadA.start();
threadB.start();
}
}

输出结果:

首先输出1-3
1
2
3
输出4-6
4
5
6
输出7-9
7
8
9
输出10-12
10
11
12

欢迎支持笔者的作品《深入理解Kafka: 核心设计与实践原理》和《RabbitMQ实战指南》,同时欢迎关注笔者的微信公众号:朱小厮的博客(ID: hiddenkafka)。
本文作者: 朱小厮

评论

Your browser is out-of-date!

Update your browser to view this website correctly. Update my browser now

×