2019独角兽企业重金招聘Python工程师标准>>>

在hadoop底层代码中，会看到有BlockingQueue的使用。

作为了解配置调优的作用，我也来了解一下BlockingQueue的使用。

BlockingQueue的核心方法：
放入数据：
　　offer(anObject):表示如果可能的话,将anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,
　　　　则返回true,否则返回false.（本方法不阻塞当前执行方法的线程）
　　offer(E o, long timeout, TimeUnit unit),可以设定等待的时间，如果在指定的时间内，还不能往队列中
　　　　加入BlockingQueue，则返回失败。
　　put(anObject):把anObject加到BlockingQueue里,如果BlockQueue没有空间,则调用此方法的线程被阻断
　　　　直到BlockingQueue里面有空间再继续.
获取数据：
　　poll(time):取走BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,
　　　　取不到时返回null;
　　poll(long timeout, TimeUnit unit)：从BlockingQueue取出一个队首的对象，如果在指定时间内，
　　　　队列一旦有数据可取，则立即返回队列中的数据。否则知道时间超时还没有数据可取，返回失败。
　　take():取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到
　　　　BlockingQueue有新的数据被加入; 
　　drainTo():一次性从BlockingQueue获取所有可用的数据对象（还可以指定获取数据的个数）， 
　　　　通过该方法，可以提升获取数据效率；不需要多次分批加锁或释放锁。

package com.test.concurrent;import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;public class Producer implements Runnable {private volatile boolean isRunning = true;private BlockingQueue queue;private static AtomicInteger count = new AtomicInteger();private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;public Producer(BlockingQueue queue) {this.queue = queue;}@Overridepublic void run() {String data = null;Random r = new Random();System.out.println("启动生产者线程！");try {while (isRunning) {System.out.println("正在生产数据...");Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));data = "data:" + count.incrementAndGet();System.out.println("将数据：" + data + "放入队列...");if (!queue.offer(data, 2, TimeUnit.SECONDS)) {System.out.println("放入数据失败：" + data);}}}catch (InterruptedException e) {e.printStackTrace();Thread.currentThread().interrupt();} finally {System.out.println("退出生产者线程！");}}public void stop() {isRunning = false;}
}

package com.test.concurrent;import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;public class Consumer implements Runnable{private BlockingQueue<String> queue;private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;public Consumer(BlockingQueue<String> queue) {this.queue = queue;}@Overridepublic void run() {System.out.println("启动消费者线程！");Random r = new Random();boolean isRunning = true;try {while (isRunning) {System.out.println("正从队列获取数据...");String data = queue.poll(2, TimeUnit.SECONDS);if (null != data) {System.out.println("拿到数据：" + data);System.out.println("正在消费数据：" + data);Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));} else {// 超过2s还没数据，认为所有生产线程都已经退出，自动退出消费线程。isRunning = false;}}}catch (InterruptedException e) {e.printStackTrace();Thread.currentThread().interrupt();} finally {System.out.println("退出消费者线程！");}}
}

package com.test.concurrent;import java.util.concurrent.*;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;public class BlockingQueueTest {public static void main(String[] args) throws InterruptedException {// 声明一个容量为10的缓存队列BlockingQueue<String> queue = new LinkedBlockingQueue<String>(10);Producer producer1 = new Producer(queue);Producer producer2 = new Producer(queue);Producer producer3 = new Producer(queue);Consumer consumer = new Consumer(queue);// 借助ExecutorsExecutorService service = Executors.newCachedThreadPool();// 启动线程service.execute(producer1);service.execute(producer2);service.execute(producer3);service.execute(consumer);// 执行10sThread.sleep(10 * 1000);producer1.stop();producer2.stop();producer3.stop();Thread.sleep(2000);// 退出Executorservice.shutdown();}
}