Java并发：任务之间协作

多个任务同时访问共享资源，是通过互斥锁来解决的。然而在任务协作时，关键问题是这些任务之间的握手。为了实现握手，使用了相同的基础特性：互斥。互斥能保证只有一个任务可以响应某个信号，这样就可以根除任务可能的竞争条件。在互斥的基础上，为任务再添加一种途径，可以将自身挂起，直到某些外部条件发生变化，表示是时候让这个任务继续向前了。

wait() 与 notify(), notifyAll()

wait() 使当前任务等待某个条件发生变化，而这个条件通常是由另一个任务来改变。wait() 会在等待外部世界产生变化的时候将任务挂起，并且只有在 notify() 或 notifyAll() 发生时，这个任务才会被唤醒并去检查所产生的变化。

wait(), notify(), notifyAll() 方法都是基类 Object 的方法，而不是属于 Thread 类的一部分，因为这些方法操作的锁是所有对象锁。

虽然这些方法是操作锁的，但是实际上，这些方法不是直接操作锁的，而只能在同步控制方法或同步控制块里调用 wait(), notify() 和 notifyAll()(由于 sleep() 不操作锁，因此可以在非同步控制方法里调用)。如果在非同步控制方法或同步控制块里调用这些方法，程序能通过编译，但是运行的时候，将得到 IllegalMonitorStateException 异常，并伴随一些含糊的消息。其实意思就是调用 wait()、notify() 和 notifyAll() 的任务在调用这些方法前必须获取对象的锁。

sleep() 与 wait() 区别

调用 sleep() 时候，锁并没有释放，调用 yield() 也属于这种情况。

然而，当一个任务在方法里遇到对 wait() 的调用的时候，线程将挂起，对象上的锁被释放。

由于 wait() 将线程挂起的同时也释放锁，这就意味着另一个任务可以获得这个锁，因此在该对象中的其他 synchronized 方法可以在 wait() 期间被其它线程调用。因此可以把 wait() 看作是 “我(当前任务)因为条件不满足而挂起了，所以我释放了这个对象的锁，其它的任务可以获取这个锁了”。

wait() 有两种形式，一种是接受毫秒数作为参数，即表示挂起一段时间，如果时间到了，从 wait() 从恢复过来，当然也可以通过 notify(), notifyAll() 来恢复。

例子

现在有两个任务，一个是涂蜡到车上，一个是抛光。抛光任务在涂蜡任务完成之前，是不能执行其它工作的，而涂蜡任务在涂另一层蜡之前，也必须等待抛光任务完成。

先定义一个 Car 类，它有涂蜡和抛光的方法

public class Car {
    private boolean waxOn = false;

    /**
     * 涂蜡
     */
    public synchronized void waxed() {
        waxOn = true; // ready to buff
        notifyAll();
    }

    /**
     * 抛光
     */
    public synchronized void buffed() {
        waxOn = false;
        notifyAll();
    }

    /**
     * 等待涂蜡
     * @throws InterruptedException
     */
    public synchronized void waitForWaxing() throws InterruptedException {
        while (!waxOn) {
            wait();
        }
    }

    /**
     * 等待抛光
     * @throws InterruptedException
     */
    public synchronized void waitForBuffing() throws InterruptedException {
        while (waxOn) {
            wait();
        }
    }

}

注意，wait(), notify(), notifyAll() 方法都是在同步方法中调用的，因为要先获得锁，才能操作这些方法。

然后定义两个任务(Runnable)，WaxOn 是用来执行涂蜡，WaxOff 用来执行抛光

public class WaxOn implements Runnable {
    private Car mCar;

    public WaxOn(Car car) {
        mCar = car;
    }

    @Override
    public void run() {
        try {
            // 线程如果被中断过，就清除中断标记，然后退出 run() 方法
            while (!Thread.interrupted()) {
                System.out.println("Wax on!");
                // 模拟涂蜡过程
                TimeUnit.MILLISECONDS.sleep(200);
                // 涂蜡完成，通知相应条件发生改变，使其它线程恢复执行
                mCar.waxed();
                // 然后，等待抛光完成时，这个时候线程挂起
                mCar.waitForBuffing();
            }
        } catch (InterruptedException e) {
            System.out.println("Exiting via interrupt");
        }
        System.out.println("Ending Wax On task");
    }
}

public class WaxOff implements Runnable {
    private Car mCar;

    public WaxOff(Car car) {
        mCar = car;
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                // 先等待涂蜡，线程挂起，如果涂蜡完成，线程恢复执行
                mCar.waitForWaxing();
                System.out.println("Wax Off!");
                // 模拟抛光过程
                TimeUnit.MILLISECONDS.sleep(200);
                // 完成抛光，通知挂起的线程继续执行
                mCar.buffed();
            }
        } catch (InterruptedException e) {
            System.out.println("Exiting via interrupt");
        }
        System.out.println("Ending Wax Off task");
    }
}

然后，多线程执行涂蜡和抛光的任务

        Car car = new Car();
        ExecutorService executorService = Executors.newCachedThreadPool();
        executorService.execute(new WaxOn(car));
        executorService.execute(new WaxOff(car));
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        executorService.shutdownNow();

执行结果如下

System.out: Wax on!
System.out: Wax Off!
...
System.out: Wax on!
System.out: Wax Off!
System.out: Wax on!
System.out: Exiting via interrupt
System.out: Ending Wax Off task
System.out: Exiting via interrupt
System.out: Ending Wax On task

这个例子，强调你必须用一个检查感兴趣的条件的 while 循环包围 wait()，这很重要，因为:

你可能有多个任务出于相同的原因在等待同一个锁，而第一个被唤醒的任务，从wait()处继续执行任务的时候，又改变了这个条件，因此我们需要调用 while() 循环来判断是继续挂起，还是执行其它的任务。
在这个任务从其 wait() 中被唤醒的时刻，有可能会有某个其他的任务已经做出了改变，从而使得这个任务在此时不能执行，或者执行其操作已经显得无关紧要。此时，应该通过 while 循环再次挂起线程，直到条件再次满足。
也有可能某些任务出于不同的原因在等待你的对象上的锁(这种情况下必须再次使用 notifyAll())。在这种情况下，你需要检查是否已经由正确的原因被唤醒，如果不是，就可以通过 while 循环再次挂起。

因此，while 循环的本质就是当线程被唤醒的时候，要再次检查条件是否满足，如果不满足就继续挂起，如果满足是执行后续的操作。

再思索下，如果 WaxOff 任务是在 WaxOn 挂起的时候才执行，那么是不是就造成了信号丢失，从而造成了死锁呢？

        try {
            final Car car = new Car();
            final ExecutorService executorService = Executors.newCachedThreadPool();
            executorService.execute(new WaxOn(car));
            new Timer().schedule(new TimerTask() {
                @Override
                public void run() {
                    executorService.execute(new WaxOff(car));
                }
            },10000);
            TimeUnit.SECONDS.sleep(20);
            executorService.shutdownNow();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

执行的结果如下

12-27 23:39:48.688 I/System.out: Wax on!
12-27 23:39:58.702 I/System.out: Wax Off!
12-27 23:39:58.905 I/System.out: Wax on!
12-27 23:39:59.109 I/System.out: Wax Off!
12-27 23:39:59.316 I/System.out: Wax on!
12-27 23:39:59.521 I/System.out: Wax Off!
12-27 23:39:59.726 I/System.out: Wax on!
12-27 23:39:59.931 I/System.out: Wax Off!
...
12-27 23:40:08.142 I/System.out: Wax Off!
12-27 23:40:08.345 I/System.out: Wax on!
12-27 23:40:08.551 I/System.out: Wax Off!
12-27 23:40:08.695 I/System.out: Exiting via interrupt
12-27 23:40:08.696 I/System.out: Exiting via interrupt
12-27 23:40:08.697 I/System.out: Ending Wax Off task
12-27 23:40:08.701 I/System.out: Ending Wax On task

从 Log 看，WaxOn 任务是在 48 秒的时候执行的，而 WaxOff 任务是在 58 秒才执行的，中间相隔 10s，也就是说 WaxOn 任务先用 notifyAll() 发出信号，然后挂起。在 10s 后， WaxOff() 任务才开始执行，然而信号并没有丢失，WaxOff 正常的执行了。

错失信号

当两个线程使用 notify()/wait() 或 notifyAll()/wait() 进行协作时，如果代码写得不严谨，可能会造成错失信号。把上面例子中 Car 类的一个方法改变下

    public  void waitForWaxing() throws InterruptedException {
        while (!waxOn) {
            // 休眠 5s，代表线程调试器切换
            Thread.sleep(5000);
            synchronized (this) {
                wait();
            }
        }
    }

wait() 的调用从同步方法改到了同步代码块中调用，而且把执行条件放到了同步块代码之外。那么当 WaxOff 任务执行的时候，会因为 Thread.Sleep(5000) 产生线程调度，那么此时 WaxOn 任务执行，把 waxOn 条件设置为了 true，当再次回到 WaxOff 任务的时候，并没有察觉到 waxOn 条件已经改变，通过 while 循环，会再次盲目地调用了 wait() 使线程挂起了。从而产生了死锁，这个时候我们只能中断线程. 而造成这个问题的根本原因就是 waxOn 变量没有加锁控制，从而没有多线程的可见性。

执行 Log 如下

12-28 00:03:33.956 I/System.out: Wax on!
12-28 00:03:43.964 I/System.out: Exiting via interrupt
12-28 00:03:43.964 I/System.out: Ending Wax Off task
12-28 00:03:43.966 I/System.out: Exiting via interrupt
12-28 00:03:43.966 I/System.out: Ending Wax On task

从 Log 中可以明显看到 WaxOff 任务并没有执行，因为出现了死锁，最终我们只能通过终结线程池来中断线程。

出现这个问题的原因是 waxOn 的修改与获取并不是线程安全的。那么，解决这个问题就很简单了，如果要访问或者修改 waxOn ，就必须先要获取对象的锁。

    public void waitForWaxing() throws InterruptedException {
        synchronized (this) {
            while (!waxOn) {
                // 休眠 2s，代表线程调试器切换
                Thread.sleep(2000);
                wait();
            }
        }
    }

这样一来，就算线程发生调度，由于还没有释放锁(Sleep不会释放锁)，因此其它线程是无法操作 waxOn 的。

使用显式的 Lock 与 Condition 对象

同步一个方法或者代码块，我们可以使用 synchronized 关键字，同时在同步代码块的时候，也可以显式的使用 Lock 对象。

线程之间的协作可以使用 Object 类的 wait(), notify()/notifyAll() 方法，也可以使用 Condition 类的 await(), signal()/signalAll() 方法。 Condition 对象由 Lock 对象的 newCondition() 方法获得。

与使用 notifyAll() 相比，signalAll() 是更安全的方式。

那么可以重写之前的 Car 类。

public class Car {
    private ReentrantLock mLock = new ReentrantLock();
    private Condition mCondition = mLock.newCondition();
    private boolean waxOn = false;

    /**
     * 涂蜡
     */
    public void waxed() {
        mLock.lock();
        try {
            waxOn = true; // ready to buff
            mCondition.signalAll();
        } finally {
            mLock.unlock();
        }

    }

    /**
     * 抛光
     */
    public void buffed() {
        mLock.lock();
        try {
            waxOn = false;
            mCondition.signalAll();
        } finally {
            mLock.unlock();
        }

    }

    /**
     * 等待涂蜡
     *
     * @throws InterruptedException
     */
    public void waitForWaxing() throws InterruptedException {
        mLock.lock();
        try {
            while (!waxOn) {
                mCondition.await();
            }
        } finally {
            mLock.unlock();
        }
    }


    /**
     * 等待抛光
     *
     * @throws InterruptedException
     */

    public void waitForBuffing() throws InterruptedException {
        mLock.lock();
        try {
            while (waxOn) {
                mCondition.await();
            }
        } finally {
            mLock.unlock();
        }
    }

}

Condition 对象是由 Lock 对象创建的
之前说到，要使用 Object 类的 wait(), notify()/notifyAll() 方法，必须先获得该对象的锁。同样，如果要使用 Condition 类的 await(), signal()/signalAll() 方法，也必须先要获得 Lock 对象的锁。
Lock 对象调用了 lock() 后，就必须要释放锁，所以后续代码块用 try-finally 包围起来，保证 Lock 对象成功释放锁。

阻塞队列

Object 类的 wait(),notifyAll()/notify() 或者 Condition 类的 await(), signalAll()/signal() 方法都是以一种非常低级的方式解决了任务互操作问题，即每次交互时都握手。在许多情况下，可以使同步队列来解决任协作问题。

同步队列在任何时刻都只允许一个任务来执行操作(插入或移除)。 java.util.concurrent.BlockingQueue 定义这个队列需要实现的接口。通常可以使用 LinkedBlockingQueue，它是一个无界队列，还可以使用 ArrayBlockingQueue，它具有固定尺寸。

如果消费者任务试图从队列中获取对象，而该队列此时为空，那么这些队列还可以挂起消费者任务，并且当有元素可用时，恢复消费者任务。

可以用 BlockingQueue 实现一个饭店仿真，服务员只需要不断的从 BlockingQueue 中获取就行了，而厨师方面，我们把情况想简单点，厨师无限的做菜，也就是无限地往 BlockingQueue 中添加元素。

public class Restaurant {

    public Restaurant() {
        ExecutorService service = Executors.newCachedThreadPool();
        BlockingQueue<Meal> mealBlockingQueue = new LinkedBlockingDeque<>();
        Chef chef = new Chef(mealBlockingQueue);
        Waiter waiter = new Waiter(mealBlockingQueue);
        service.execute(chef);
        service.execute(waiter);
        try {
            TimeUnit.MILLISECONDS.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        service.shutdownNow();
    }

}

class Chef implements Runnable {
    private BlockingQueue<Meal> mBlockingQueue;
    private int mCount;

    public Chef(BlockingQueue<Meal> mealBlockingQueue) {
        mBlockingQueue = mealBlockingQueue;
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                mBlockingQueue.put(new Meal(++mCount));
                System.out.println("Order up! " + mCount);
            }
        } catch (InterruptedException e) {
            System.out.println("Chef interrupt");
        }
    }
}

class Waiter implements Runnable {
    private BlockingQueue<Meal> mBlockingQueue;

    public Waiter(BlockingQueue<Meal> mealBlockingQueue) {
        mBlockingQueue = mealBlockingQueue;
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                Meal meal = mBlockingQueue.take();
                System.out.println("Waiter got " + meal);
            }
        } catch (InterruptedException e) {
            System.out.println("Waiter interrupt");
        }
    }
}

class Meal {
    private final int id;

    public Meal(int id) {
        this.id = id;
    }

    @Override
    public String toString() {
        return "Meal " + id;
    }
}

管道

在 BlockingQueue 之前，线程间通信可以通过管道进行输入/输出，但是相比较而方，BlockingQueue 更健壮且更容易。

class Sender implements Runnable {
    private Random mRandom = new Random(47);
    private PipedWriter mWriter = new PipedWriter();

    public PipedWriter getWriter() {
        return mWriter;
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                for (char c = 'A'; c <= 'Z'; c++) {
                    mWriter.write(c);
                    TimeUnit.MILLISECONDS.sleep(mRandom.nextInt(500));
                }
            }
        } catch (IOException e) {
            System.out.println("Sender write exception");
        } catch (InterruptedException e) {
            System.out.println("Sender sleep interrupt");
        }
    }
}

class Receiver implements Runnable {
    private PipedReader mReader;

    public Receiver(Sender sender) throws IOException{
        mReader = new PipedReader(sender.getWriter());
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                System.out.println("Read: " + (char)mReader.read() + ",");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

CountDownLatch

允许一个线程或多个线程等待，直到其它线程执行完毕才恢复执行。

CountDownLatch 在初始化的时候，给一个计数值。调用 countDown() 可以减小这个计数值，如果计数值到达0，
那么调用了 await() 方法而阻塞的线程就会恢复执行。

CountDownLatch 被设计为只触发一次，计数值不能被重置。如果需要能够重置计数值的版本，则可以使用 CyclicBarrier.

任务在调用 countDown() 的时候并没有被阻塞，只有当计数值不为0的时候，任务调用 await() 会被阻塞。

假如现在把一个单独的任务分为N个独立的小任务，并用有M个任务在等待这N个任务执行完毕后才能执行。

public class TaskPortion implements Runnable {
    private static int counter = 0;
    private final int id = counter++;
    private static Random sRandom = new Random(47);
    private final CountDownLatch mCountDownLatch;

    public TaskPortion(CountDownLatch countDownLatch) {
        mCountDownLatch = countDownLatch;
    }

    @Override
    public void run() {
        try {
            TimeUnit.MILLISECONDS.sleep(sRandom.nextInt(2000));
            System.out.println(this + "complete");
            mCountDownLatch.countDown();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public String toString() {
        return String.format(Locale.getDefault(), "%1$-3d", id);
    }
}
public class WaitingTask implements Runnable {
    private static int counter = 0;
    private final int id = counter++;
    private final CountDownLatch mCountDownLatch;

    public WaitingTask(CountDownLatch countDownLatch) {
        mCountDownLatch = countDownLatch;
    }

    @Override
    public void run() {
        try {
            mCountDownLatch.await();
            System.out.println("Latch barrier passed for " + this);
        } catch (InterruptedException e) {
            System.out.println(this + " interrupted");
        }
    }

    @Override
    public String toString() {
        return String.format(Locale.getDefault(), "WaitingTask %1$-3d", id);
    }
}

public class Test{

    public static void main(string[] args) {
        ExecutorService service = Executors.newCachedThreadPool();
        CountDownLatch latch = new CountDownLatch(100);
        for (int i = 0; i < 10; i++) {
            service.execute(new WaitingTask(latch));
        }
        for (int i = 0; i < 100; i++) {
            service.execute(new TaskPortion(latch));
        }
        System.out.println("Launched all tasks");
        service.shutdown();
    }
}

WaitingTask 任务首先会进入阻塞状态，当100个 TaskPortion 任务全部执行完毕后，10个 WaitingTask 任务全部恢复执行。

CyclicBarrier

循环栅栏，适用于这样的情况：你有N个独立的任务并行执行，这些任务都是几个相同的阶段，下进行下一个阶段前，需要等待之前
所有任务完成，因此这些任务可以一致地向前移动。这非常像 CountDownLatch ，只是 CountDownLatch 只触发一次，而 CyclicBarrier
可以重用多次。

还是利用上面的例子，假如有3个任务并行执行，在每执行下一步的时候，都必须等待所有任务都完成才能执行。

public class TaskPortion implements Runnable {
    private static int counter = 0;
    private final int id = counter++;
    private static Random sRandom = new Random(47);
    private final CyclicBarrier mCyclicBarrier;

    public TaskPortion(CyclicBarrier cyclicBarrier) {
        mCyclicBarrier = cyclicBarrier;
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                TimeUnit.MILLISECONDS.sleep(sRandom.nextInt(2000));
                System.out.println(this + "complete");
                mCyclicBarrier.await();
            }
        } catch (InterruptedException e) {
            System.out.println(this + " interrupted");
        } catch (BrokenBarrierException e) {
            System.out.println(this + " BrokenBarrierException");
        }
    }

    @Override
    public String toString() {
        return String.format(Locale.getDefault(), "%1$-3d", id);
    }
}

public class TestActivity extends AppCompatActivity {

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_test_acrtivity);
        final ExecutorService service = Executors.newCachedThreadPool();
        CyclicBarrier cyclicBarrier = new CyclicBarrier(3, new Runnable() {
            private int count;

            @Override
            public void run() {
                if (++count == 3) {
                    System.out.println("Stop!");
                    service.shutdownNow();
                } else {
                    System.out.println("Ready,go!");
                }
            }
        });
        for (int i = 0; i < 3; i++) {
            service.execute(new TaskPortion(cyclicBarrier));
        }
    }
}

结果如下

System.out: 1  complete
System.out: 2  complete
System.out: 0  complete
System.out: Ready,go!
System.out: 2  complete
System.out: 1  complete
System.out: 0  complete
System.out: Ready,go!
System.out: 2  complete
System.out: 1  complete
System.out: 0  complete
System.out: Stop!

并行执行三个 TaskPortion，每当他们执行完毕，都会等待，也就是 mCyclicBarrier.await()，当三个线程同时等待的时候，
也就代表三个任务都完成了，然后就会执行一个 Runnable，这个 Runnable 就是初始化 CyclicBarrier 的。执行完毕后，三个任务
就又恢复执行下一阶段的任务，直到三个阶段执行完毕，线程池关闭。

CyclicBarrier 的构造器也可以不传入 Runnable。而且 CyclicBarrier 还有一个 await(long timeout, TimeUnit unit) 方法。

DelayQueue

是一个无界的 BlockingQueue，也就是说线程安全，用于放置实现了 Delayed 接口的对象，其中的对象只能在其到期时，才能取走。
这种队列是有序的，即队头对象的延迟到期的时间最长。

Delayed 接口是继承自 Comparable 接口

public interface Delayed extends Comparable<Delayed> {

    /**
     * Returns the remaining delay associated with this object, in the
     * given time unit.
     *
     * @param unit the time unit
     * @return the remaining delay; zero or negative values indicate
     * that the delay has already elapsed
     */
    long getDelay(TimeUnit unit);
}

Comparable 接口是来用比较排序的，而纯粹的 Delayed 接口是用来提供截止时间的。从 getDelay() 的注释看，返回值代表剩下的时间，如果返回值小于或等于0，代表剩余时间为0.

public class DelayedTask implements Runnable, Delayed {
    private static int counter;
    private final int id = counter++;
    private final int delta;
    private final long trigger;
    protected static List<DelayedTask> sequence = new ArrayList<>();

    public DelayedTask(int delayInMilliseconds) {
        delta = delayInMilliseconds;
        trigger = System.nanoTime()
                + TimeUnit.NANOSECONDS.convert(delta, TimeUnit.MILLISECONDS);
        sequence.add(this);
    }

    @Override
    public long getDelay(@NonNull TimeUnit unit) {
        return unit.convert(trigger - System.nanoTime(), TimeUnit.NANOSECONDS);
    }

    @Override
    public int compareTo(@NonNull Delayed o) {
        DelayedTask delayedTask = (DelayedTask) o;
        if (trigger < delayedTask.trigger) return -1;
        if (trigger > delayedTask.trigger) return 1;
        return 0;
    }

    @Override
    public void run() {
        System.out.println(this + " ");
    }

    @Override
    public String toString() {
        return String.format("[%1$-4d]", delta) + " Task " + id;
    }

    public String summary() {
        return "(" + id + ":" + delta + ")";
    }

    public static class EndSentinel extends DelayedTask {
        private ExecutorService e;

        public EndSentinel(int delayInMilliseconds, ExecutorService e) {
            super(delayInMilliseconds);
            this.e = e;
        }

        @Override
        public void run() {
            for (DelayedTask pt : sequence) {
                System.out.println(pt.summary() + " ");
            }
            System.out.println(this + " calling shutdownNow()");
            e.shutdownNow();
        }
    }
}

在构造函数的参数中，提供了一个 delay 时间，然后根据现在的时间，计算出未来触发事件的时间，也就是 trigger 变量。

getDealay() 函数用于返回剩余的时间，而这里正是用触发时间 trigger 减去现在的时间。根据 getDelay() 函数的描述，如果返回值小于或等于0，那么代表时间已到。

compareTo() 的实现，从方法注释看，如果当前对象小于指定的对象，返回负值，如果等于，返回0，如果大于，返回正值。所以，这里就是这样实现的。

DelayedTask 实现了 Delayed 接口，这样 DelayedTask 对象可以添加到 DelayQueue 中，然后可以根据延迟时间点来执行。

public class DelayedTaskConsumer implements Runnable {
    private DelayQueue<DelayedTask> q;

    public DelayedTaskConsumer(DelayQueue<DelayedTask> q) {
        this.q = q;
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                q.take().run();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("Finished DelayedTaskConsumer");
    }
}

        Random random = new Random(47);
        ExecutorService service = Executors.newCachedThreadPool();
        DelayQueue<DelayedTask> queue = new DelayQueue<>();
        for (int i = 0; i < 10; i++) {
            queue.put(new DelayedTask(random.nextInt(5000)));
        }
        queue.add(new DelayedTask.EndSentinel(5000, service));
        service.execute(new DelayedTaskConsumer(queue));

PriorityBlockingQueue

可阻塞的优先级队列。加入到这个队列中的元素，需要实现 Comparable 接口，用于比较。

Semaphore

正常的锁(来自 concurrent.locks 或 synchronized 锁) 在任何时刻都只允许一个任务访问一项资源，而 Semaphore(计数信号量) 允许 n 个任务同时访问这个资源。这就好像给线程发“许可证”一样，只有拥有“许可证”的线程才可以访问某个资源。

public class Pool<T> {
    private int size;
    private List<T> items = new ArrayList<>();
    private volatile boolean[] checkedOut;
    private Semaphore available;

    public Pool(Class<T> classObject, int size) {
        this.size = size;
        checkedOut = new boolean[size];
        available = new Semaphore(size, true);
        try {
            for (int i = 0; i < size; i++) {
                items.add(classObject.newInstance());
            }
        } catch (InstantiationException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        }
    }

    public T checkOut() throws InterruptedException {
        // 如果获取不到信号量，就会阻塞线程
        available.acquire();
        return getItem();
    }

    private synchronized T getItem() {
        for (int i = 0; i < size; i++) {
            if (!checkedOut[i]) {
                checkedOut[i] = true;
                return items.get(i);
            }
        }
        return null;
    }

    public void checkIn(T x) {
        if (releaseItem(x)) {
            available.release();
        }
    }

    private synchronized boolean releaseItem(T item) {
        int index = items.indexOf(item);
        if (index == -1) {
            return false;
        }
        if (checkedOut[index]) {
            checkedOut[index] = false;
            return true;
        }
        return false;
    }
}

Pool 用来管理对象集，checkOut() 方法会先调用 Semaphore 对象的 acqurire() 方法，来获取锁，如果锁的计数为0，将阻塞线程。之后通过一个同步方法 getItem() 来获取对象。从这里可以看出，Semaphore 只是限制了同时访问资源的数量，然而它并没有起到同步的效果，还是要用 synchronized 的同步方法来获取对象。

checkIn() 方法首先通过同步方法 releaseItem() 来确认是否可以签回，注意到，同样使用到同步方法。如果确认可以签回，就调用 Semaphore 对象的 release() 来释放“许可证”。从这里也可以看出，Semaphore 实际只是起到控制并发访问资源的数量，而并没有起到同步效果。同步问题，还是得用 lock 或者 synchronized 来实现。

Exchanger

Exchanger 是在两个任务之间交换对象的栅栏。当这些任务进入栅栏时，它们各自拥有一个对象，当它们离开时，它们都拥有之前由对象持有的对象。 Exchanger 的典型应用场景是：一个任务在创建对象，这些对象的生产代价很高昂，所以需要另外一个任务马上消费掉才能继续创建下一个对象。

public class ExchangerProduce implements Runnable {
    private List<Integer> mIntegers;
    private Exchanger<List<Integer>> mListExchanger;
    private int count;

    public ExchangerProduce(List<Integer> integers, Exchanger<List<Integer>> listExchanger) {
        mIntegers = integers;
        mListExchanger = listExchanger;
    }

    @Override
    public void run() {
        try {
            for (int i = 0; i < 5; i++) {
                mIntegers.add(i);
            }
            mIntegers = mListExchanger.exchange(mIntegers);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("Producer: " + mIntegers);
    }
}

在 run() 方法中，Exchanger 的 exchange() 方法把参数 mIntegers 数据交出去，返回的 mIntegers 是换回来的数据。

public class ExchangerConsumer implements Runnable {
    private List<Integer> mIntegers;
    private Exchanger<List<Integer>> mListExchanger;

    public ExchangerConsumer(List<Integer> integers, Exchanger<List<Integer>> listExchanger) {
        mIntegers = integers;
        mListExchanger = listExchanger;
    }

    @Override
    public void run() {
        try {
            for (int i = 5; i < 10; i++) {
                mIntegers.add(i);
            }
            mIntegers = mListExchanger.exchange(mIntegers);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("Consumer: " + mIntegers);
    }
}

在 run() 方法中，也是换出了参数 mIntegers 的数据，并返回了换回的数据，保存在 mIntegers 中。

如果并发启动两个任务

        ExecutorService service = Executors.newCachedThreadPool();
        List<Integer> producerList = new CopyOnWriteArrayList<>();
        List<Integer> consumerList = new CopyOnWriteArrayList<>();
        Exchanger<List<Integer>> exchanger = new Exchanger<>();
        service.execute(new ExchangerProduce(producerList, exchanger));
        service.execute(new ExchangerConsumer(consumerList, exchanger));