java多线程基本概述(二十一)——BlockingQueue

public interface BlockingQueue<E>extends Queue<E>

支持两个附加操作的 Queue,这两个操作是:获取元素时等待队列变为非空,以及存储元素时等待空间变得可用。

BlockingQueue 方法以四种形式出现,对于不能立即满足但可能在将来某一时刻可以满足的操作,这四种形式的处理方式不同:第一种是抛出一个异常,第二种是返回一个特殊值(null 或 false,具体取决于操作),第三种是在操作可以成功前,无限期地阻塞当前线程,第四种是在放弃前只在给定的最大时间限制内阻塞。下表中总结了这些方法: 

       抛出异常       特殊值       阻塞      超时
插入    add(e)       offer(e)    put(e)    offer(e, time, unit)
移除    remove()     poll()      take()    poll(time, unit)
检查    element()    peek()      不可用     不可用

BlockingQueue 不接受 null 元素。试图 addput 或 offer 一个 null 元素时,某些实现会抛出 NullPointerExceptionnull 被用作指示 poll 操作失败的警戒值。

BlockingQueue 可以是限定容量的。它在任意给定时间都可以有一个 remainingCapacity,超出此容量,便无法无阻塞地 put 附加元素。没有任何内部容量约束的 BlockingQueue 总是报告 Integer.MAX_VALUE 的剩余容量。

BlockingQueue 实现主要用于生产者-使用者队列,但它另外还支持 Collection 接口。因此,举例来说,使用 remove(x) 从队列中移除任意一个元素是有可能的。然而,这种操作通常 会有效执行,只能有计划地偶尔使用,比如在取消排队信息时。

BlockingQueue 实现是线程安全的。所有排队方法都可以使用内部锁或其他形式的并发控制来自动达到它们的目的。然而,大量的 Collection 操作(addAllcontainsAllretainAll 和 removeAll没有 必要自动执行,除非在实现中特别说明。因此,举例来说,在只添加了 c 中的一些元素后,addAll(c) 有可能失败(抛出一个异常)。

BlockingQueue 实质上 支持使用任何一种“close”或“shutdown”操作来指示不再添加任何项。这种功能的需求和使用有依赖于实现的倾向。例如,一种常用的策略是:对于生产者,插入特殊的 end-of-stream 或 poison 对象,并根据使用者获取这些对象的时间来对它们进行解释。

以下是基于典型的生产者-使用者场景的一个用例。注意,BlockingQueue 可以安全地与多个生产者和多个使用者一起使用。

 class Producer implements Runnable {
   private final BlockingQueue queue;
   Producer(BlockingQueue q) { queue = q; }
   public void run() {
     try {
       while(true) { queue.put(produce()); }
     } catch (InterruptedException ex) { ... handle ...}
   }
   Object produce() { ... }
 }

 class Consumer implements Runnable {
   private final BlockingQueue queue;
   Consumer(BlockingQueue q) { queue = q; }
   public void run() {
     try {
       while(true) { consume(queue.take()); }
     } catch (InterruptedException ex) { ... handle ...}
   }
   void consume(Object x) { ... }
 }

 class Setup {
   void main() {
     BlockingQueue q = new SomeQueueImplementation();
     Producer p = new Producer(q);
     Consumer c1 = new Consumer(q);
     Consumer c2 = new Consumer(q);
     new Thread(p).start();
     new Thread(c1).start();
     new Thread(c2).start();
   }
 }
例子: LiftOff.java
package tij;

/**
 * Created by huaox on 2017/3/31.
 *
 */
public class LiftOff implements Runnable {

    protected int countDown = 10;
    private static int taskDown = 0;
    private final int id = taskDown++;

    public LiftOff() {

    }
    public LiftOff(int countDown) {
        this.countDown = countDown;
    }

    public String status(){
        return "【#"+id+"("+countDown+")】";
    }
    @Override
    public void run() {
        try {
            while (countDown-->0){
                //Thread.sleep(5000);
                System.out.println("liftoff= "+status()+" was   used  "+System.currentTimeMillis());
                Thread.yield();
            }
        } catch (Exception e) {
            System.out.println("interrupted !!!!liftOff");
        }

    }

}

Test.java

package tij;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.concurrent.*;

/**
 * Created by huaox on 2017/3/31.
 *
 */
 class RunnerBlockingQueue implements Runnable {
    private BlockingQueue<LiftOff> rockets;
    RunnerBlockingQueue(BlockingQueue<LiftOff> rockets) {
        this.rockets = rockets;
    }

    void add(LiftOff liftOff){
        try {
            System.out.println("liftoff= "+liftOff.status()+/*"queue size="+rockets.size()+*/" put  start  "+System.currentTimeMillis());
            rockets.put(liftOff);
            //System.out.println("liftoff= "+liftOff.status()+" put    end  "+System.currentTimeMillis());
        }catch (InterruptedException e){
            System.out.println("interrupting during put()");
        }
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()){
//                System.out.println("liftoff   get "+System.currentTimeMillis());
                LiftOff liftOff = rockets.take();
                System.out.println("liftoff= "+liftOff.status()+" start  run  "+System.currentTimeMillis());   //line 1
                liftOff.run();
                //System.out.println("liftoff= "+liftOff.status()+"  run  end    "+System.currentTimeMillis()); //line 2
            }
        }catch (InterruptedException e){
            System.out.println("waking from take()");
        }
        System.out.println("Exiting runner");
    }
}

public class Test{

     static void getKey(String msg){
         System.out.println(msg);
         try {
             String result = new BufferedReader(new InputStreamReader(System.in)).readLine();
             System.out.println("finish block io "+result);
         }catch (IOException e){
             throw new RuntimeException(e);
         }
     }
     static void test(String msg,BlockingQueue<LiftOff> queue) throws InterruptedException {
         System.out.println("using queue test"+msg);
         RunnerBlockingQueue runner = new RunnerBlockingQueue(queue);
         Thread thread = new Thread(runner);
         thread.start();//这时候开始的时候,队列里面并没有任务。那么取队列任务运行的时候会被阻塞
         TimeUnit.SECONDS.sleep(1);
         for (int i = 0; i < 5; i++) {
             runner.add(new LiftOff(5));
         }
         getKey("Press 'Enter'("+msg+")");//阻塞直到控制台输入后才发送中断消息
         thread.interrupt();
         System.out.println("Finished "+msg+" test");
     }

    public static void main(String[] args) throws InterruptedException {
        test("LinkedBlockingDeque",new LinkedBlockingDeque<LiftOff>()); //未限制大小
//        test("ArrayBlockingQueue",new ArrayBlockingQueue<LiftOff>(3)); //限制大小为3
//        test("SynchronousQueue",new SynchronousQueue<>()); //大小为1
    }
}

输出结果:

using queue testLinkedBlockingDeque
liftoff= 【#0(5)】 put  start  1492681266001    put 0
liftoff= 【#1(5)】 put  start  1492681266001    put 1
liftoff= 【#0(5)】 start  run  1492681266001             start use 0  
liftoff= 【#2(5)】 put  start  1492681266001    put 2
liftoff= 【#0(4)】 was   used  1492681266001             use 0
liftoff= 【#3(5)】 put  start  1492681266001    put 3
liftoff= 【#0(3)】 was   used  1492681266001        use 0
liftoff= 【#4(5)】 put  start  1492681266001    put 4
liftoff= 【#0(2)】 was   used  1492681266001        use 0
Press 'Enter'(LinkedBlockingDeque)
liftoff= 【#0(1)】 was   used  1492681266001        use 0
liftoff= 【#0(0)】 was   used  1492681266001        use 0
liftoff= 【#1(5)】 start  run  1492681266001                    start use 1
liftoff= 【#1(4)】 was   used  1492681266001                  use   1
liftoff= 【#1(3)】 was   used  1492681266001                            use  1 
liftoff= 【#1(2)】 was   used  1492681266001                 use  1 
liftoff= 【#1(1)】 was   used  1492681266001                 use  1 
liftoff= 【#1(0)】 was   used  1492681266001                  use  1
liftoff= 【#2(5)】 start  run  1492681266001                        start use 2
liftoff= 【#2(4)】 was   used  1492681266001                        use 2
liftoff= 【#2(3)】 was   used  1492681266001                        use 2
liftoff= 【#2(2)】 was   used  1492681266001                        use 2
liftoff= 【#2(1)】 was   used  1492681266001                        use 2
liftoff= 【#2(0)】 was   used  1492681266001                        use 2
liftoff= 【#3(5)】 start  run  1492681266001                                  start use 3
liftoff= 【#3(4)】 was   used  1492681266001                                  use 3
liftoff= 【#3(3)】 was   used  1492681266001                                  use 3
liftoff= 【#3(2)】 was   used  1492681266001                                  use 3
liftoff= 【#3(1)】 was   used  1492681266001                                  use 3
liftoff= 【#3(0)】 was   used  1492681266001                                  use 3
liftoff= 【#4(5)】 start  run  1492681266001
liftoff= 【#4(4)】 was   used  1492681266001
liftoff= 【#4(3)】 was   used  1492681266001
liftoff= 【#4(2)】 was   used  1492681266001
liftoff= 【#4(1)】 was   used  1492681266001
liftoff= 【#4(0)】 was   used  1492681266001
finish Task
finish block io finish Task
Finished LinkedBlockingDeque test
waking from take()
Exiting runner

Process finished with exit code 0
图中结果状态描绘的很清除了,相信一看就能明白吧!
如果把Task也就是任务具体的运行者,LiftOff的run方法里面的加个休眠。
 @Override
    public void run() {
        try {
            while (countDown-->0){
                Thread.sleep(500);
                System.out.println("liftoff= "+status()+" was   used  "+System.currentTimeMillis());
                Thread.yield();
            }
        } catch (Exception e) {
            System.out.println("interrupted !!!!liftOff");
        }

    }

那么运行结果:

using queue testLinkedBlockingDeque
liftoff= 【#0(5)】 put  start  1492682392952        put 0
liftoff= 【#0(5)】 start  run  1492682392952        start 0
liftoff= 【#1(5)】 put  start  1492682392952        put 1
liftoff= 【#2(5)】 put  start  1492682392952        put 2
liftoff= 【#3(5)】 put  start  1492682392952        put 3
liftoff= 【#4(5)】 put  start  1492682392952        put 4
Press 'Enter'(LinkedBlockingDeque)
liftoff= 【#0(4)】 was   used  1492682393452              use 0
liftoff= 【#0(3)】 was   used  1492682393962              use 0
liftoff= 【#0(2)】 was   used  1492682394462              use 0
liftoff= 【#0(1)】 was   used  1492682394962              use 0
liftoff= 【#0(0)】 was   used  1492682395478              use 0
liftoff= 【#1(5)】 start  run  1492682395478                  start use 1
liftoff= 【#1(4)】 was   used  1492682395978                  use 1
liftoff= 【#1(3)】 was   used  1492682396485                  use 1
liftoff= 【#1(2)】 was   used  1492682396995                  use 1
liftoff= 【#1(1)】 was   used  1492682397496                  use 1
liftoff= 【#1(0)】 was   used  1492682397996                  use 1
liftoff= 【#2(5)】 start  run  1492682397996                      start use 2  
liftoff= 【#2(4)】 was   used  1492682398496                      use 2    
liftoff= 【#2(3)】 was   used  1492682399012                      use 2  
liftoff= 【#2(2)】 was   used  1492682399512                      use 2
liftoff= 【#2(1)】 was   used  1492682400027                      use 2  
liftoff= 【#2(0)】 was   used  1492682400528                      use 2  
liftoff= 【#3(5)】 start  run  1492682400528
liftoff= 【#3(4)】 was   used  1492682401043
liftoff= 【#3(3)】 was   used  1492682401543
liftoff= 【#3(2)】 was   used  1492682402043
liftoff= 【#3(1)】 was   used  1492682402543
liftoff= 【#3(0)】 was   used  1492682403057
liftoff= 【#4(5)】 start  run  1492682403057
liftoff= 【#4(4)】 was   used  1492682403557
liftoff= 【#4(3)】 was   used  1492682404057
liftoff= 【#4(2)】 was   used  1492682404558
liftoff= 【#4(1)】 was   used  1492682405058
liftoff= 【#4(0)】 was   used  1492682405558
exit
finish block io exit
Finished LinkedBlockingDeque test
waking from take()
Exiting runner

Process finished with exit code 0

 LinkedBlockingDeque 可以允许同时多个任务添加进来,但是取的时候会一个接着一个来。

下面测试:把line 1注释掉,line 2取消注释。并且打开下面代码的注释
test("ArrayBlockingQueue",new ArrayBlockingQueue<LiftOff>(3)); //限制大小为3

输出结果:

using queue testArrayBlockingQueue
liftoff= 【#0(5)】 put    end  1492683884689  put       queue size =1
liftoff= 【#0(5)】 start  run  1492683884689 take      queue size =0 
liftoff= 【#1(5)】 put    end  1492683884689 put      queue size =1 
liftoff= 【#2(5)】 put    end  1492683884689 put      queue size =2 
liftoff= 【#3(5)】 put    end  1492683884689 put      queue size =3 
liftoff= 【#0(4)】 was   used  1492683885190
liftoff= 【#0(3)】 was   used  1492683885690
liftoff= 【#0(2)】 was   used  1492683886205
liftoff= 【#0(1)】 was   used  1492683886705
liftoff= 【#0(0)】 was   used  1492683887206    此刻。0将运行完成,将运行下一个任务,队列长度变为2
liftoff= 【#1(5)】 start  run  1492683887206    任务1开始运行
liftoff= 【#4(5)】 put    end  1492683887206 put 于此同时,添加新的任务到队列中来
Press 'Enter'(ArrayBlockingQueue)
liftoff= 【#1(4)】 was   used  1492683887721
liftoff= 【#1(3)】 was   used  1492683888221
liftoff= 【#1(2)】 was   used  1492683888721
liftoff= 【#1(1)】 was   used  1492683889227
liftoff= 【#1(0)】 was   used  1492683889727
liftoff= 【#2(5)】 start  run  1492683889727
liftoff= 【#2(4)】 was   used  1492683890227
liftoff= 【#2(3)】 was   used  1492683890743
liftoff= 【#2(2)】 was   used  1492683891243
liftoff= 【#2(1)】 was   used  1492683891743
liftoff= 【#2(0)】 was   used  1492683892243
liftoff= 【#3(5)】 start  run  1492683892243
liftoff= 【#3(4)】 was   used  1492683892743
liftoff= 【#3(3)】 was   used  1492683893243
liftoff= 【#3(2)】 was   used  1492683893759
liftoff= 【#3(1)】 was   used  1492683894259
liftoff= 【#3(0)】 was   used  1492683894759
liftoff= 【#4(5)】 start  run  1492683894759
liftoff= 【#4(4)】 was   used  1492683895259
liftoff= 【#4(3)】 was   used  1492683895759
liftoff= 【#4(2)】 was   used  1492683896271
liftoff= 【#4(1)】 was   used  1492683896787
liftoff= 【#4(0)】 was   used  1492683897287

Process finished with exit code 1

 ArrayBlockingQueue 一个有界队列,大小固定,那么队列中同时允许只有大小的任务,只有任务出队后才能添加新任务

下面测试:
test("SynchronousQueue",new SynchronousQueue<>())

输出结果:

using queue testSynchronousQueue
liftoff= 【#0(5)】 start  run  1492685180178
liftoff= 【#0(4)】 put    end  1492685180178      put 0
liftoff= 【#0(4)】 was   used  1492685180679
liftoff= 【#0(3)】 was   used  1492685181179
liftoff= 【#0(2)】 was   used  1492685181690      虽然此过程中取出了,任务队列中数量为0.但也要最新出队列的任务完成后,才能接收下个任务。这是一个保证,否则也就没意思了
liftoff= 【#0(1)】 was   used  1492685182205
liftoff= 【#0(0)】 was   used  1492685182705
liftoff= 【#1(5)】 start  run  1492685182705
liftoff= 【#1(5)】 put    end  1492685182705      put 1
liftoff= 【#1(4)】 was   used  1492685183206
liftoff= 【#1(3)】 was   used  1492685183706
liftoff= 【#1(2)】 was   used  1492685184206
liftoff= 【#1(1)】 was   used  1492685184706
liftoff= 【#1(0)】 was   used  1492685185206
liftoff= 【#2(5)】 start  run  1492685185206
liftoff= 【#2(5)】 put    end  1492685185206      put 2
liftoff= 【#2(4)】 was   used  1492685185706
liftoff= 【#2(3)】 was   used  1492685186221
liftoff= 【#2(2)】 was   used  1492685186721
liftoff= 【#2(1)】 was   used  1492685187222
liftoff= 【#2(0)】 was   used  1492685187724
liftoff= 【#3(5)】 put    end  1492685187724      put 3
liftoff= 【#3(5)】 start  run  1492685187724
liftoff= 【#3(4)】 was   used  1492685188239
liftoff= 【#3(3)】 was   used  1492685188740
liftoff= 【#3(2)】 was   used  1492685189240
liftoff= 【#3(1)】 was   used  1492685189741
liftoff= 【#3(0)】 was   used  1492685190241
liftoff= 【#4(5)】 start  run  1492685190241
liftoff= 【#4(5)】 put    end  1492685190241      put 4
Press 'Enter'(SynchronousQueue)
liftoff= 【#4(4)】 was   used  1492685190742
liftoff= 【#4(3)】 was   used  1492685191256
liftoff= 【#4(2)】 was   used  1492685191762
liftoff= 【#4(1)】 was   used  1492685192274
liftoff= 【#4(0)】 was   used  1492685192774

Process finished with exit code 1

可见 SynchronousQueue 一次只允许一个任务进行。



原文地址:https://www.cnblogs.com/soar-hu/p/6740284.html