时间戳

首页 > 技术文章 > java学习之浅谈多线程3--线程间协作

java学习之浅谈多线程3--线程间协作

nangch 2015-10-21 12:36 原文

通过保证临界区上多个线程的相互排斥，线程同步完全可以避免竞争状态的发生，但是有时还需要线程之间的协作。有两种方式可用于线程间的通信。
1.使用条件Condition
Condition接口：

1 +await():void           当前线程等待直到发生某个条件
2 +signal():void          唤醒一个等待线程
3 +signalAll():Condition  唤醒所有等待线程

条件是通过调用Lock对象的newCondition()方法而创建的对象，为了使用条件，必须首先获取锁，await()方法让线程等待并且自动释放条件上的锁，一旦条件正确，线程重新获取锁并且继续执行。创建了条件，该对象就可以使用await(),signal()和signalAll()方法。

考虑典型的消费者/生产者例子。假设使用缓冲区储存整数。缓冲区的大小是受限的。缓冲区提供write(int)方法将一个int值添加到缓冲区中，还提供方法read()从缓冲区中读取和删除一个int值为了同步这个操作，使用具有两个条件的锁：notEmpty(即缓冲区非空)和notFull(即缓冲区未满)。当任务向缓冲区添加一个int时，如果缓冲区是满的，那么任务将会等待notFull状态。当任务从缓冲区中删除一个int时，如果缓冲区是空的，那么任务将等待notEmpty状态。代码如下：

  1 import java.util.LinkedList;
  2 import java.util.concurrent.ExecutorService;
  3 import java.util.concurrent.Executors;
  4 import java.util.concurrent.locks.Condition;
  5 import java.util.concurrent.locks.Lock;
  6 import java.util.concurrent.locks.ReentrantLock;
  7 
  8 public class ConsumerProducer {
  9     private static Buffer buffer = new Buffer();    //创建缓冲区
 10 
 11     public static void main(String[] args) {
 12         //创建线程池
 13         ExecutorService executor = Executors.newFixedThreadPool(2);
 14 
 15         executor.execute(new ProducerTask());
 16         executor.execute(new ConsumerTask());
 17 
 18         executor.shutdown();
 19     }
 20 
 21     private static class ProducerTask implements Runnable {
 22         @Override
 23         public void run() {
 24             try {
 25                 int i = 1;
 26                 while(true) {
 27                     System.out.println("Producer writes " + i) ;
 28                     buffer.write(i++);
 29                     Thread.sleep((int)(Math.random() * 10000));
 30                 }
 31             } catch (InterruptedException e) {
 32             }
 33         }
 34     }
 35 
 36     private static class ConsumerTask implements Runnable {
 37         @Override
 38         public void run() {
 39             try {
 40                 while(true) {
 41                     System.out.println("\t\t\tConsumer reads " + buffer.read());
 42                     Thread.sleep((int)(Math.random() * 10000));
 43                 }
 44             } catch (InterruptedException e) {
 45 
 46             }
 47         }
 48     }
 49 
 50     /**
 51      * 创建内部类Buffer作为缓冲区
 52      * @author zx
 53      *
 54      */
 55     private static class Buffer {
 56         private static final int CAPACITY = 1;  //缓冲区的大小
 57         private LinkedList<Integer> queue = new LinkedList<Integer>();  
 58 
 59         private static Lock lock = new ReentrantLock();     //创建一个锁
 60 
 61         //创建两个条件
 62         private static Condition notEmpty = lock.newCondition();    //缓冲区非空
 63         private static Condition notFull = lock.newCondition();     //缓冲区未满
 64 
 65         public void write(int value) {
 66             lock.lock();        //获得锁
 67 
 68             try {
 69                 while(queue.size() == CAPACITY) {
 70                     System.out.println("Wait for notFull condition");
 71                     notFull.await();    //暂停当前线程，等待notFull条件
 72                 }
 73 
 74                 queue.offer(value);     //将整数value加入缓冲区
 75                 notEmpty.signal();      //唤醒等待notEmpty条件的线程
 76             } catch (InterruptedException e) {
 77             } finally {
 78                 lock.unlock();          //释放锁
 79             }
 80         }
 81 
 82         @SuppressWarnings("finally")
 83         public int read() {
 84             int value = 0;
 85             lock.lock();                //获得锁
 86             try {
 87                 while(queue.isEmpty()) {
 88                     System.out.println("\t\tWait for notEmpty condition");
 89                     notEmpty.await();   //暂停线程等待notEmpty条件
 90                 }
 91 
 92                 value = queue.remove(); //从缓冲区删除一个整数
 93                 notFull.signal();       //唤醒等待notFull条件的线程
 94             } catch (InterruptedException e) {
 95             } finally {
 96                 lock.unlock();          //释放锁
 97                 return value;
 98             }
 99         }
100 
101     }
102 }

2.使用阻塞队列

阻塞队列在试图向一个满队列添加元素或者从空队列中删除元素时会导致线程阻塞。BlockingQueue接口扩展java.util.Queue,并且提供同步的put和take方法向队列头添加元素，以及从队列尾删除元素。

Java支持三个具体的阻塞队列ArrayBlockingQueue、LinkedBlockingQueue和PriorityBlockingQueue。
ArrayBlockingQueue使用数组实现阻塞队列。必须指定一个容量或者可选的公平性来构造ArrayBlockingQueue。
LinkedBlockingQueue使用链表实现阻塞队列。可以创建不受限的或者受限的LinkedBlockingQueue。
PriorityBlockingQueue是优先队列，可以创建不受限的或者受限的优先队列。

下面的程序使用ArrayBlockingQueue来简化上面的生产者/消费者程序。创建一个ArrayBlockingQueue来存储整数，生产者线程将一个整数放入队列中，而消费者线程从队列中取走一个整数。

 1 import java.util.concurrent.ArrayBlockingQueue;
 2 import java.util.concurrent.ExecutorService;
 3 import java.util.concurrent.Executors;
 4 
 5 public class ConsumerProducerUsingBlockingQueue {
 6     //创建buffer用来存储整数
 7     private static ArrayBlockingQueue<Integer> buffer = 
 8             new ArrayBlockingQueue<Integer>(2);
 9 
10     public static void main(String[] args) {
11         //创建线程池
12         ExecutorService executor = Executors.newFixedThreadPool(2);
13 
14         executor.execute(new ProducerTask());
15         executor.execute(new ConsumerTask());
16 
17         executor.shutdown();
18     }
19 
20     private static class ProducerTask implements Runnable {
21         @Override
22         public void run() {
23             try {
24                 int i = 1;
25                 while(true) {
26                     System.out.println("Producer writes " + i);
27                     buffer.put(i++);        //向队列添加元素
28                     Thread.sleep((int)(Math.random() * 10000));
29                 }
30             } catch (InterruptedException e) {
31             }
32         }
33     }
34 
35     private static class ConsumerTask implements Runnable {
36         @Override
37         public void run() {
38             try {
39                 while(true) {
40                     //从队列删除元素
41                     System.out.println("\t\t\tConsumer reads " + buffer.take());
42                     Thread.sleep((int)(Math.random() * 10000));
43                 }
44             } catch (InterruptedException e) {
45             }
46         }
47     }
48 }

推荐阅读