阻塞队列

傷城~ 2022-05-27 05:48 201阅读 0赞

### 阻塞队列 ###

阻塞队列：最常见的业务场景就是生产者不断生产任务放进阻塞队列中，消费者不断从阻塞队列中获取任务；当阻塞队列中填满数据时，所有生产者端的线程自动阻塞，当阻塞队列中数据为空时，所有消费端的线程自动阻塞。这些操作BlockingQueue都已经包办了，不用我们程序员去操心了。

阻塞队列我们常用的有：LinkedBlockingQueue和ArrayBlockingQueue，它们在各方面还是很大的区别的；ArrayBlockingQueue在put,take操作使用了同一个锁，两者操作不能同时进行，而LinkedBlockingQueue使用了不同的锁，put操作和take操作可同时进行，以此来提高整个队列的并发性能。

作为开发者，使用阻塞队列需要注意的一点是：如果构造一个LinkedBlockingQueue对象，而没有指定其容量大小，LinkedBlockingQueue会默认一个类似无限大小的容量（Integer.MAX\_VALUE），这样的话，如果生产者的速度一旦大于消费者的速度，也许还没有等到队列满阻塞产生，系统内存就有可能已被消耗殆尽了。

### 阻塞队列的一些常用方法 ###

![41e100e99fb3a46a8bdfdffd6def2457.png][]

package queue.demo;
    
    import java.util.Random;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.LinkedBlockingQueue;
    
    /**
     * @author liuchaojun
     * @date 2018-4-12 下午05:15:14
     */
    public class BlockQueueDemo {
    	public static void main(String[] args) {
    		BlockingQueue<Integer> queue = new LinkedBlockingQueue<Integer>(2); // 定长为2的阻塞队列
    		// ExecutorService：真正的线程池接口
    		ExecutorService service = Executors.newCachedThreadPool(); // 缓存线程池
    		// 创建3个生产者：
    		ProducerDemo p1 = new ProducerDemo("车鉴定web端", queue);
    		ProducerDemo p2 = new ProducerDemo("车鉴定APP端", queue);
    		ProducerDemo p3 = new ProducerDemo("车鉴定APP端", queue);
    		// 创建三个消费者：
    		ConsumerDemo c1 = new ConsumerDemo("消费者001", queue);
    		ConsumerDemo c2 = new ConsumerDemo("消费者002", queue);
    		ConsumerDemo c3 = new ConsumerDemo("消费者003", queue);
    
    		// 启动线程
    		service.execute(p1);
    		service.execute(p2);
    		service.execute(p3);
    		service.execute(c1);
    		service.execute(c2);
    		service.execute(c3);
    
    	}
    }
    
    /**
     * 生产者
     */
    class ProducerDemo implements Runnable {
    	private String producerName;
    	private BlockingQueue queue;// 阻塞队列
    	private Random r = new Random();
    
    	// 构造函数,传入生产者名称和操作的阻塞队列
    	public ProducerDemo(String producerName, BlockingQueue queue) {
    		this.producerName = producerName;
    		this.queue = queue;
    	}
    
    	@Override
    	public void run() {
    		while (true) {
    			try {
    				int task = r.nextInt(100); // 产生随机数
    				System.out.println(producerName + "开始生产任务：" + task);
    				queue.put(task); // 生产者向队列中放入一个随机数
    				Thread.sleep(1000); // 减缓生产者生产的速度，如果队列为空，消费者就会阻塞不会进行消费直到有数据被生产出来
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    }
    
    class ConsumerDemo implements Runnable {
    	private String consumerName;
    	private BlockingQueue queue;// 阻塞队列
    
    	// 构造函数,传入消费者名称和操作的阻塞队列
    	public ConsumerDemo(String consumerName, BlockingQueue queue) {
    		this.consumerName = consumerName;
    		this.queue = queue;
    	}
    
    	@Override
    	public void run() {
    		while (true) {
    			try {
    				System.out.println(consumerName + "开始消费任务---" + queue.take());// 消费者从阻塞队列中消费一个随机数
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    }

[41e100e99fb3a46a8bdfdffd6def2457.png]: /images/20220527/8d3336901871423b9b94e3592d129dd8.png