Java -- 使用阻塞队列(BlockingQueue)控制线程通信

BlockingQueeu接口是Queue的子接口,但是它的主要作用并不是作为容器,而是作为线程同步的工具。

特征:

       当生产者线程试图向BlockingQueue中放入元素时,如果该队列已满,则该线程被阻塞;当消费者线程试图从BlockingQueue中取出元素时,如果该队列已空,则该线程被阻塞。

程序的两个线程通过交替从BlockingQueue中放入元素、取出元素,即可很好地控制线程的通信。

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

       put(E e):尝试从BlockingQueue的头部取出元素,如果该队列的元素已满,则阻塞该线程。

       take():尝试从BlockingQueue的头部取出元素,如果该队列的元素已空,则阻塞该线程。

BlockingQueue继承了Queue接口,当然也可以使用Queue接口的方法。这些方法归纳起来可分为3组:

  • 在队列尾部插入元素。包括add(E e)、offer(E e)和put(E e)方法,当该队列已满时,这3个方法分别抛出异常、返回false、阻塞队列。
  • 在队列头部删除并返回删除的元素。包括remove()、poll()和take()方法。当该队列已空时,这3个方法分别抛出异常、返回false、阻塞队列。
  • 在队列头部取出但是不删除元素。包括element()和peek()方法,当队列已空时,这两个方法分别抛出异常、返回false。

BlockingQueue队列的5个实现类:

  • ArrayBlockingQueue:基于数组实现的BlockingQueue队列
  • LinkedBlockingQueue:基于链表实现的BlockingQueue队列
  • PriorityBlockingQueue:.....
  • SynchronousQueue:同步队列。对对垒的存、取操作必须交替进行
  • DelayQueue:.....

具体代码示例:

package com.fuchenggang.blockingqueue;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/** 
* @ClassName: Producer 
* @Description: 生产者
* @author fucg@kunyitech.com 
* @date 2015年7月10日 下午4:08:03  
*/ 
class Producer extends Thread {
    private BlockingQueue<String> bq;

    public Producer(BlockingQueue<String> bq) {
        this.bq = bq;
    }

    @Override
    public void run() {
        String[] strArr = new String[]
        {
            "Java",
            "Struts",
            "Spring"
        };
        
        for (int i = 0; i < 999999999; i++) {
            System.out.println(getName() + "生产者准备生产集合元素!");
            
            try {
                Thread.sleep(200);
                //尝试放入元素,如果队列已满,则线程被阻塞
                bq.put(strArr[i%3]);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            
            System.out.println(getName() + "生产完成:" +bq);
        }
    }
}

/** 
* @ClassName: Consumer 
* @Description: 消费者
* @author fucg@kunyitech.com 
* @date 2015年7月10日 下午4:08:19  
*/ 
class Consumer extends Thread{

    private BlockingQueue<String> bq;
    
    public Consumer(BlockingQueue<String> bq) {
        this.bq = bq;
    }
    
    @Override
    public void run() {
        while (true) {
            System.out.println(getName() + "消费者准备消费集合元素");
            
            try {
                Thread.sleep(200);
                //尝试取出元素,如果队列已空,则线程被阻塞
                bq.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            
            System.out.println(getName() + "消费完成:" + bq);
        }
    }
}

/** 
* @ClassName: BlockingQueueTest2 
* @Description: 测试生产者线程和消费者线程的通信
* @author fucg@kunyitech.com 
* @date 2015年7月10日 下午4:13:37  
*/ 
public class BlockingQueueTest2 {
    public static void main(String[] args) {
        //创建一个容器为1的BlockingQueue
        BlockingQueue<String> bq = new ArrayBlockingQueue<String>(1);
        
        //启动3个生产者线程
        new Producer(bq).start();
        new Producer(bq).start();
        new Producer(bq).start();
        
        //启动1个消费者线程
        new Consumer(bq).start();
    }
}

 代码讲解:

程序启动了3个生产者线程向BolckingQueue集合放入元素,启动了1个消费者线程从BlockingQueue集合中取出元素。本程序的BlockingQueue集合容量为1,因此3个生产者线程无法连续放入元素,必须等待消费者线程取出一个元素后,3个生产者线程之一才能放入一个元素。

运行结果:

原文地址:https://www.cnblogs.com/chengxuyuanzhilu/p/chengxuyuanzhilu.html