BlockQueue生产者消费者例子

BlockingQueue详解

阻塞队列，顾名思义，首先它是一个队列，
    
常用的队列主要有以下两种：
        先进先出（FIFO）：先插入的队列的元素也最先出队列，类似于排队的功能。
        从某种程度上来说这种队列也体现了一种公平性。
　　 后进先出（LIFO）：后插入队列的元素最先出队列，这种队列优先处理最近发生的事件。

阻塞队列的核心就是生产者和消费着
        当队列中没有数据的情况下，消费者端的所有线程都会被自动阻塞（挂起），直到有数据放入队列
        队列中填满数据的情况下，生产者端的所有线程都会被自动阻塞（挂起），直到队列中有空的位置，线程被自动唤醒。

    
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取出一个队首的对象，如果在指定时间内，队列一旦有数据可取，则立即返回队列中的数据。否则知道时间超时还没有数据可取，返回null。
    　　    take():取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到BlockingQueue有新的数据被加入; 
    　　    drainTo():一次性从BlockingQueue获取所有可用的数据对象（还可以指定获取数据的个数），通过该方法，可以提升获取数据效率；不需要多次分批加锁或释放锁。

 

列子：

 

                package test;  

  

                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;  

                    }  

                      

                    @Override  

                    public 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();  

                                if(!queue.offer(data, 2, TimeUnit.SECONDS)){  

                                    System.out.println("放入数据失败："+data);  

                                    continue;  

                                }  

                                  

                                System.out.println("将数据：" + data + "放入队列...");  

                                  

                            }  

                        }catch(InterruptedException e){  

                            e.printStackTrace();  

                            Thread.currentThread().interrupt();  

                        }finally{  

                            System.out.println("====退出生产者线程====");  

                        }  

                    }  

                      

                    public void stop(){  

                        isRunning = false;  

                    }  

                      

                }  

  

                package test;  

  

                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;  

                    }  

  

                    @Override  

                    public 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{  

                                    isRunning = false;  

                                }  

                            }  

                        }catch(InterruptedException e){  

                            e.printStackTrace();  

                            Thread.currentThread().interrupt();  

                        }finally{  

                            System.out.println("====退出消费者线程！====");  

                        }  

                    }  

                      

                }  

  

                package test;  

  

                import java.util.concurrent.BlockingQueue;  

                import java.util.concurrent.ExecutorService;  

                import java.util.concurrent.Executors;  

                import java.util.concurrent.LinkedBlockingQueue;  

  

  

                public class Test {  

  

                    public static void main(String[] args) throws InterruptedException {  

                        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);  

                          

                        ExecutorService service = Executors.newCachedThreadPool();  

                        service.execute(producer1);  

                    service.execute(producer2);  

                    service.execute(producer3);  

                    service.execute(consumer);  

  

                    Thread.sleep(10 * 1000); //生产者执行10m  

                    producer1.stop();  

                    producer2.stop();  

                    producer3.stop();  

                      

                    Thread.sleep(2000);  

                      

                    //退出Executor  

                    service.shutdown();  

                    }  

                }