Java5 多线程(三)--Lock和Condition实现线程同步通信

1>Lock: Lock比传统线程模型中的Synchronied方式更加面向对象,与生活中的锁类似,锁本身也应该是一个对象.两个线程执行的代码段要实现同步互斥的效果,它们必须用同一个Lock对象,锁是在代表要操作的资源的类的内部方法中,而不是线程代码中. 上面的输出器Outputer类就可以这样改写:class Outputer2 { // 声明一个锁 Lock lock = new ReentrantLock(); public void print(String name) { //把要互斥的代码写在lock()和unlock()方法之间 lock.lock(); try { for (int i = 0; i < name.length(); i++) { System.out.print(name.charAt(i)); } System.out.println();// 打印完字符串换行 } finally{ //如果中途抛出异常,那么这把锁就没有被解锁,别人就进不来了 //所以写在finally里面 lock.unlock(); } }} 读写锁,分为读锁和写锁,多个读锁不互斥,读锁和写锁互斥,写锁与写锁互斥,这是JVM自己控制的,你只要上好相应的锁即可,如果你的代码只读数据,可以很多人同时读,但不能同时写,那就上读锁;如果你的代码修改数据,只能有一个人在写,且不能同时读取,那就上写锁.总之,读的时候上读锁,写的时候上写锁! 看如下程序: 新建6个线程,3个线程用来读,3个线程用来写,final Queue3 q3 = new Queue3(); for (int i = 0; i < 3; i++) { new Thread() { public void run() { while (true) { q3.get(); } } }.start(); new Thread() { public void run() { while (true) { q3.put(new Random().nextInt(10000)); } } }.start(); }然后在编写一个类Queue3 里面有一个读方法和写方法:class Queue3 { private Object data = null;// 共享数据，只能有一个线程能写该数据，但可以有多个线程同时读该数据。 //读写锁 ReadWriteLock rwl = new ReentrantReadWriteLock(); // 相当于读操作 public void get() { rwl.readLock().lock(); try { System.out.println(Thread.currentThread().getName() + " be ready to read data!"); Thread.sleep((long) (Math.random() * 1000)); System.out.println(Thread.currentThread().getName() + "have read data :" + data); } catch (InterruptedException e) { e.printStackTrace(); } finally { rwl.readLock().unlock(); } } // 相当于写操作 public void put(Object data) { rwl.writeLock().lock(); try { System.out.println(Thread.currentThread().getName() + " be ready to write data!"); Thread.sleep((long) (Math.random() * 1000)); this.data = data; System.out.println(Thread.currentThread().getName() + " have write data: " + data); } catch (InterruptedException e) { e.printStackTrace(); } finally { rwl.writeLock().unlock(); } }}这样可以实现正常的逻辑,如果我们把读写锁相关的代码注释,发现程序正准备写的时候,就有线程读了,发现准备读的时候,有线程去写,这样不符合我们的逻辑通过Java5的新特新可以很轻松的解决这样的问题.查看Java API ReentrantReadWriteLock 上面有经典(缓存)的用法.这是上面的伪代码.class CachedData {
Object data;
volatile boolean cacheValid;
final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();

void processCachedData() {
rwl.readLock().lock();
if (!cacheValid) {
// Must release read lock before acquiring write lock
rwl.readLock().unlock();
rwl.writeLock().lock();
try {
// Recheck state because another thread might have
// acquired write lock and changed state before we did.
if (!cacheValid) {
data = ...
cacheValid = true;
}
// Downgrade by acquiring read lock before releasing write lock
rwl.readLock().lock();
} finally {
rwl.writeLock().unlock(); // Unlock write, still hold read
}
}
try {
use(data);
} finally {
rwl.readLock().unlock();
}
}
}基于上面的例子,我们可以实现一个缓存系统. Map<String, Object> cache = new HashMap<String, Object>();ReadWriteLock rrwl = new ReentrantReadWriteLock(); public Object getData(String key) { rrwl.readLock().lock(); Object value = null; try { value = cache.get(key); if (value == null) { rrwl.readLock().unlock(); rrwl.writeLock().lock(); try { //假设三个线程同时去获取写锁,我们知道只有第一个线程能够获取 //那么其他两个线程只有等了,如果第一个线程按流程执行完后,刚才的两个线程可以得到写锁了, //然后接着就可以修改数据了(赋值).所以加上判断! if (value == null) {//为什么还要在这里判断一次.? value = "hello world"; } // 降级,通过释放写锁之前获取读锁 rrwl.readLock().lock(); } finally { rrwl.writeLock().unlock(); } } } finally { rrwl.readLock().unlock(); } return value; }
2>Condition: Condition的功能类似于在传统的线程技术中的,Object.wait()和Object.notify()的功能,在等待Condition时,允许发生"虚假唤醒",这通常作为对基础平台语义的让步,对于大多数应用程序,这带来的实际影响很小,因为Condition应该总是在一个循环中被等待,并测试正被等待的状态声明.某个实现可以随意移除可能的虚假唤醒,但是建议程序员总是假定这些虚假唤醒可能发生,因此总是在一个循环中等待. 一个锁内部可以有多个Condition,即有多路等待和通知,可以参看jdk1.5提供的Lock与Condition实现的可阻塞队列的应用案例,从中要体味算法,还要体味面向对象的封装.在传统的线程机制中,一个监视器对象上只能有一路等待和通知,要想实现多路等待和通知,必须嵌套使用多个同步监视器对象.(如果只用一个Condition,两个放的都在等,一旦一个放进去了,那么它会通知可能会导致另一个放的接着往下走). 我们也可以通过Lock和Condition来实现上面我们讲的例子:子线程循环10次，接着主线程循环100，接着又回到子线程循环10次，接着再回到主线程又循环100，如此循环50次，请写出程序。这里只实现其中的一个方法,因为其他的是一样的.Lock lock = new ReentrantLock(); Condition condition = lock.newCondition(); // 默认子线程先执行 boolean isShouldSub = true; public void sub(int k) { lock.lock();//相当于synchronied try { while(!isShouldSub) { try { condition.await();//然后实现通信 } catch (InterruptedException e) { e.printStackTrace(); } } for (int i = 1; i <= 10; i++) { System.out .println("sub thread sequence " + i + " loop of " + k); } // 子线程做完了,把它置为false isShouldSub = false; // 并且唤醒主线程 condition.signal(); } finally { lock.unlock(); } } 可以使用Lock和Condition来实现一个缓冲队列(要区别缓冲和缓存的区别),其实jdk api有这样的例子,如下: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;//如果putptr已经是数组的最后一个,那么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呢?用一个Condition似乎也能实现. public void put(Object x) throws InterruptedException { //如果有5个线程执行到此方法里面,那么只有一个线程获取到锁
lock.lock(); // 锁住了别的线程就不能进来了,包括下面的take()因为他们用的是同一把锁
try { //如果已经放满
while (count == items.length)
notFull.await();//执行到此,锁就释放了,可能这里就有5个线程在此等,其他线程就可以调用take()方法去取了然后调用signal()然而5个线程中,只有一个线程能被唤醒.该被唤醒的线程执行到signal时候,唤醒其他线程.如果用一个Condition,唤醒的可能就是上面的4个线程,而这个4个线程是往里面(put),而应该唤醒的是去取(take())线程.因为已经放满了如果再通知线程去放,那么就出现逻辑错误了.所以这里用到两个Condition的妙处!
items[putptr] = x;
if (++putptr == items.length) putptr = 0;
++count;
notEmpty.signal();
} finally {
lock.unlock();
}
}
据此,我们可以改变子线程循环10次，接着主线程循环100，接着又回到子线程循环10次，接着再回到主线程又循环100，如此循环50次，请写出程序。的例子,这个例子是两个线程之间的跳转,那么如果实现三个线程之间的轮循,比如:线程1循环10,线程2循环100,线程3循环20次,然后又是线程1,接着线程2...一直轮循50次.class Business3 { Lock lock = new ReentrantLock(); Condition sub1 = lock.newCondition(); Condition sub2 = lock.newCondition(); Condition sub3 = lock.newCondition(); //默认线程1执行 int shouldSub = 1; public void sub1(int k) { lock.lock();//相当于synchronied try { while (shouldSub!=1) { try { sub1.await();//然后实现通信 } catch (InterruptedException e) { e.printStackTrace(); } } for (int i = 1; i <= 10; i++) { System.out .println("sub1 thread sequence " + i + " loop of " + k); } // 把值置为2,然线程2可执行 shouldSub = 2; // 线程1做完后,只唤醒线程2 sub2.signal(); } finally { lock.unlock(); } } public void sub2(int k) { lock.lock(); try { while (shouldSub!=2) { try { sub2.await(); } catch (InterruptedException e) { e.printStackTrace(); } } for (int i = 1; i <= 100; i++) { System.out.println("sub2 thread sequence " + i + " loop of " + k); } // 把值置为3,然线程3可执行 shouldSub = 3; // 线程2做完后,只唤醒线程3 sub3.signal(); } finally { lock.unlock(); } } public void sub3(int k) { lock.lock(); try { while (shouldSub!=3) { try { sub3.await(); } catch (InterruptedException e) { e.printStackTrace(); } } for (int i = 1; i <= 20; i++) { System.out.println("sub3 thread sequence " + i + " loop of " + k); } // 把值置为1,然线程1可执行 shouldSub = 1; // 线程3做完后,只唤醒线程1 sub1.signal(); } finally { lock.unlock(); } }
转载请注明出处: http://blog.csdn.net/johnny901114/article/details/8695708