Java提高——JUC线程Condition控制通信通信、阻塞队列

布满荆棘的人生 2022-05-20 02:21 233阅读 0赞

##### Condition控制线程通信 #####

如果程序中不适用synchronized关键字来保证线程同步，而直接使用Lock对象来保证线程同步，则系统中不存在隐式的同步监视器，也就不能使用wait()、notify()、notifyAll()方法进行线程通信了。

使用Lock对象来保证线程同步时，Java提供了一个Condition类来保持协调——使用Condition类可以让已经得到Lock对象的那些无法继续执行的线程释放Lock对象，Condition对象也可以唤醒其他处于等待的线程。

Condition对象中提供的3个方法：

1）await()：类似wait()，导致当前线程等待，直到其他线程调用该Condition的signal(),signalAll()方法来唤醒该线程。

2）signal()：唤醒在此Lock对象上等待的一个线程

3）signalAll()：唤醒在此Lock对象上等待的所有线程。

##### 阻塞队列（BlockingQueue）控制线程通信 #####

BlockingQueue作为线程同步的工具，具有一个特征：当生产者线程试图向BlockingQueue中放入元素时，如果该队列已满，则线程阻塞；当消费者试图从BlockingQueue取出元素时，如果该队列为空，则该线程被阻塞。

BlockingQueue提供如下两个支持阻塞的方法：

1、put（E e）:把e放入BlockingQueue中，如果队列满，则线程阻塞；

2、take（）：尝试从BlockingQueue中取出元素，如果队列空，则线程阻塞。

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td>抛出异常</td> 
   <td>返回false</td> 
   <td>阻塞线程</td> 
   <td>指定超时时长</td> 
  </tr> 
  <tr> 
   <td>对尾插入元素&nbsp; （队列满时）</td> 
   <td>add（e）</td> 
   <td>offer（）</td> 
   <td>put（e）</td> 
   <td>offer（e，time，unit）</td> 
  </tr> 
  <tr> 
   <td>对头删除元素（队列空时）</td> 
   <td>remove（）</td> 
   <td>poll（）</td> 
   <td>塔克（）</td> 
   <td>poll（time，unit）</td> 
  </tr> 
  <tr> 
   <td>获取、不删除元素</td> 
   <td>element（）</td> 
   <td>peek（）</td> 
   <td>无</td> 
   <td>无</td> 
  </tr> 
 </tbody> 
</table>

![70][]

ArrayBlockingQueue：基于数组实现的BlockingQueue队列

LinkedBlockingQueue:基于链表实现的BlockingQueue队列

PriorityBlockingQueue：不是标准的阻塞队列。

SynchronousQueue：同步队列，对该队列的存取操作必须交替进行。

DelayQueue：是一个特殊的BlockingQueue，底层基于PriorityBlockingQueue实现，不过DelayQueue要求集合元素都实现Delay接口。DelayQueue根据集合元素的getDelay()方法返回值进行排序。

public class BlockingQueueTest {
        public static void main(String[] args) throws InterruptedException {
            //定义一个长度为2的阻塞队列
            BlockingQueue blockingQueue = new ArrayBlockingQueue(2);
            blockingQueue.put("java");
            blockingQueue.put("c++");
            blockingQueue.put("python");//线程阻塞，如果使用add将会抛出异常，使用offer将会返回false
        }
    }

class Producers extends Thread{
        private BlockingQueue blockingQueue;
    
        public Producers(BlockingQueue blockingQueue) {
            this.blockingQueue = blockingQueue;
        }
        @Override
        public void run(){
            try {
                String[] str = new String[]{
      "java","c","net"};
                for (int i = 0; i < 999999; i++) {
                    System.out.println(getName()+"生产者准备要生产元素");
                    Thread.sleep(200);
                    //尝试放入元素，如果队列满了，则阻塞
                    blockingQueue.put(str[i%3]);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(getName()+"生产完成");
        }
    }
    class Consumers extends Thread{
        private BlockingQueue blockingQueue;
    
        public Consumers(BlockingQueue blockingQueue) {
            this.blockingQueue = blockingQueue;
        }
        @Override
        public void run(){
            try {
                while (true){
                    System.out.println(getName()+"从队列中取出元素");
                    Thread.sleep(200);
                    //如果队列空了则阻塞
                    blockingQueue.take();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(getName()+"消费完成");
        }
    }
    public class BlockingQueueTest {
        public static void main(String[] args) throws InterruptedException {
           /* //定义一个长度为2的阻塞队列
            BlockingQueue blockingQueue = new ArrayBlockingQueue(2);
            blockingQueue.put("java");
            blockingQueue.put("c++");
            blockingQueue.put("python");//线程阻塞，如果使用add将会抛出异常，使用offer将会返回false*/
    
            //创建一个容量为1的BlockingQueue
            BlockingQueue blockingQueue = new ArrayBlockingQueue(1);
            //启动三个生产者线程
            new Producers(blockingQueue).start();
            new Producers(blockingQueue).start();
            new Producers(blockingQueue).start();
            //启动一个消费者线程
            new Consumers(blockingQueue).start();
    
        }
    }

程序启动了3个线程，但是BlockingQueue中只有一个空间，因此三个生产者无法连续放入元素，只有等消费者消费了之后三个生产者之一才能够生产一个元素并放入。

[70]: /images/20220520/cacc64d99d8644e0a89b7dbd178d5bf9.png