《Java编程思想》——并发读书笔记

自食其力，多疑而自信。

速度的提高是以多核处理器的形式而不是更快的芯片的形式出现的。为了使程序运行的更快，你必须学习如何利用这些额外的处理器，而这正是并发赋予你的能力。

强有力的多处理器web服务器的常见情况就是在这些处理器之间之间分布多个任务，从而可以极大地提高吞吐量。在为每个请求分配一个线程的程序中，它可以将大量的用户请求分布到多个CPU上。
但是，并发通常是提高运行在单处理器上的程序的性能。
- 解释：当程序有一行代码发生阻塞（通常是IO）时，没有并发的程序就会整个停下来，而有并发的程序的多个任务就会继续向前执行，所以没有阻塞，没有并发的程序性能高于并发程序（因为并发程序需要切换任务）。
编写多线程程序最基本的困难在协调不同线程驱动的任务之间对内存和IO这种资源的使用，以使得这些资源不会同时被多个任务访问。
Java的线程机制是抢占式的，这表示调度机制会周期性地中断线程，将上下文切换到另一个线程，从而为每隔县城都提供时间片，使得每个线程都会分配到数量合理的时间去驱动它的任务。在协作多线程中，每个任务都会自动的放弃控制，这要求程序员要有意识地在每个任务中插入某种类型的让步语句。
协作式系统的优点：上下文切换的开销通常比抢占式系统要低廉许多，并且对可以同时执行的线程数量在理论上没有任何限制。
了解并发利于掌握基于消息机制的架构。
单个的线程可以拥有多个并发执行的任务。

2 基本的线程机制

2.1定义任务

LearnConcurrency.java

public class LearnConcurrency implements Runnable {
        protected int countDown = ;
        private static int taskCount = ;
    //    final初始化后，不希望被修改
        private final int id = taskCount++;
        public LearnConcurrency(){}
        public LearnConcurrency(int countDown) {
            this.countDown = countDown;
        }
        public String status() {
            return "#" + id + "(" + (countDown > ?countDown :"Liftoff!") +"),";
        }

        public void run() {
            while (countDown-- > ) {
                System.out.print(status());
    //            对线程调度器建议：我已经执行完生命周期中最重要的部分了，此刻正是切换给其他任务执行一段时间的大好时机
                Thread.yield();
            }
        }
    }

任务的run()方法通常总会有某种形式的循环，是的任务一直运行下去知道不再需要，所以要设定跳出循环的条件。通常，run()被写成无限循环的形式，这就意味着，除非有某个条件使得run()终止，否则他将永远运行下去。
Thread.yield() 对线程调度器建议：我已经执行完生命周期中最重要的部分了，此刻正是切换给其他任务执行一段时间的大好时机

2.2 Thread类

要实现线程行为，必须显式地将一个任务附着到线程上，传统方式是把它提交给Thread构造器。

```java

    public class BasicThread {
        public static void main(String[] args) {
            Thread thread = new Thread(new LearnConcurrency());
            thread.start();
            System.out.println("Waiting for LiftOff");
        }
    }
```

上面的代码运行后有两个线程两个方法在运行，main和run。
线程调度机制是非确定性的。
垃圾回收对于Thread对象，在其任务完成退出run()并且死亡之前，GC无法对其清楚。
一个线程创建一个单独的执行线程，在对start()的调用完成之后，他仍旧会继续存在。

2.3 使用Executor

```java
    public class CachedThreadPool {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 5; i++) {
            executorService.execute(new LearnConcurrency());
        }
        executorService.shutdown();
    }

```

经常使用单个的Executor创建和管理系统中所有的任务。

```java

        public class CachedThreadPool {
            public static void main(String[] args) {
                ExecutorService executorService = Executors.newFixedThreadPool(5);
                for (int i = 0; i < 5; i++) {
                    executorService.execute(new LearnConcurrency());
                }
                executorService.shutdown();
            }
        }

```

FixedThreadPool可以一次性预先执行代价高昂的线程分配，限制了线程数量。不用为每个任务都固定的付出创建线程的开销，所以省时间。
限制线程数量的好处在于防止线程的滥用。

SingleThreadExecutor用于希望在另一个线程中连续运行的任何事物（长期存活的任务）。例如监听进入的套接字连接的任务（他只有一个线程）。

```java

public class SingleThreadExecutor {
        public static void main(String[] args) {
        ExecutorService executorService =
            Executors.newSingleThreadExecutor();
        for (int i = 0; i < 5; i++) {
            executorService.execute(new LearnConcurrency());
        }
        executorService.shutdown();
    }
}

```

如果向SingleThreadExecutor提交了多个任务，由于其内部维护了一个隐藏的悬挂任务队列，他会按照提交的顺序执行，不会发生交叉。
假设你有大量线程，且他们都需要使用文件系统，可以使用SingleThreadExecutor来运行这些线程，以确保在任意时刻只有一个任务在运行，而不需要处理同步问题和不会产生对文件系统的过度使用。

2.4 从任务中产生返回值

需要实现Callable接口，而不是Runnable接口，需要使用ExecutorService.submit()方法调用它

它的类型参数表示的是从方法call（）（而不是run()）中返回的值。

```java


public class TaskWithResult implements Callable<String> {
    private int id;

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

    @Override
    public String call() throws Exception {
        return "result of TaskWithResult " + id;
    }
}






@Test
public void test1() {
    ExecutorService executorService = Executors.newCachedThreadPool();
    ArrayList<Future<String>> results =
            new ArrayList<>();
    for (int i = 0; i < 10; i++) {
        results.add(executorService.submit(new TaskWithResult(i)));
    }
    for (Future<String> future:
            results) {
        try {
            System.out.println(future.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } finally {
            executorService.shutdown();
        }
    }
}

```

其中 executorService.submit(new TaskWithResult()); 返回 future 对象，Futrue可以监视目标线程调用call的情况，当你调用Future的get()方法以获得结果时，当前线程就开始阻塞，直到call方法结束返回结果。
可以用isDown查询Future是否已经完成。只有在调用get的时候，当前线程才开始阻塞。

2.5 休眠

在run中产生的异常不能跨线程传播回main()，所以必须在本地处理所有在任务内部产生的异常。
sleep会使得线程睡眠（即阻塞），这使得线程调度器可以切换到另一个线程，进而驱动另一个任务。

2.6 优先级

调度器将倾向于优先权更高的线程先执行（执行的频率高），但CPU处理线程集的顺序还是不确定的。
可以通过Thread.currentThread().setPriority(int)设置优先级，getPriority()获取优先级。
JDK有10个优先级。
Windows有7个优先级。
大部分情况下不要操控优先级。

2.7 让步

对于任何重要的控制或在调整应用时，都不能依赖于yield()。

2.8 后台线程

是指在程序运行的时候在后台提供的一种通用服务的线程，不属于程序中不可或缺的部分。
当所有非后台线程结束时，程序也就终止了，同时会杀死进程中的所有后台线程。

main()就是一个非后台线程。

```java
public class SimpleDaemons implements Runnable {
    @Override
    public void run() {
        try {
            while (true) {
                TimeUnit.MICROSECONDS.sleep(1000);
                System.out.println(Thread.currentThread() + " " + this);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        for (int i = 0; i < 10; i++) {
            Thread daemon = new Thread(new SimpleDaemons());
            daemon.setDaemon(true);
            daemon.start();//必须在线程启动之前设置为后台线程
        }
        System.out.println("main over");
    }
}
```

结果

```

Thread[Thread-4,5,main] [email protected]
Thread[Thread-0,5,main] [email protected]
Thread[Thread-2,5,main] [email protected]
Thread[Thread-1,5,main] [email protected]
main over
Thread[Thread-1,5,main] [email protected]
Thread[Thread-2,5,main] [email protected]
Thread[Thread-4,5,main] [email protected]
Thread[Thread-0,5,main] [email protected]
Thread[Thread-1,5,main] [email protected]
Thread[Thread-2,5,main] [email protected]
Thread[Thread-4,5,main] [email protected]
Thread[Thread-0,5,main] [email protected]
Thread[Thread-4,5,main] [email protected]
Thread[Thread-0,5,main] [email protected]
Thread[Thread-6,5,main] [email protected]
Thread[Thread-2,5,main] [email protected]


```

一个后台线程产生工厂

```java

public class DaemonThreadFactory implements ThreadFactory {
    @Override
    public Thread newThread(Runnable r) {
        Thread thread = new Thread(r);
        thread.setDaemon(true);
        return thread;
    }
}

```

finally子句根本不会执行，但是如果去掉setDaemon()的调用，就会看到finally子句将会执行。

```java

class ADaemon implements Runnable {
    @Override
    public void run() {
        try {
            System.out.println("Starting ADemon");
            TimeUnit.MICROSECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            System.out.println("run finally");
        }
    }

}
public class DaemonsDontRunFinally {
    public static void main(String[] args) {
        Thread thread = new Thread(new ADaemon());
        thread.setDaemon(true);
        thread.start();
    }
}

```

这是因为当最后一个非后台线程终止后，后台线程会突然终止。
意味着在main()在子线程睡着的1ms的时候已经关闭了，这时候JVM会立即关闭所有的后台进程，而不会有任何的确认形式。

2.9 编码的变体

管用的自管理的Runnable()

public class SelfManaged implements Runnable {

        private int countDown = ;
        private Thread thread = new Thread(this);

        public SelfManaged() {
            thread.start();
        }

        @Override
        public String toString() {
            return Thread.currentThread().getName() + "(" +countDown + "),";
        }

        @Override
        public void run() {
            while (true) {
                System.out.println(this);
                if (--countDown == ) {
                    return;
                }
            }
        }
    }

通过这种方式可以继承另一个不同的类（如果需要的话）。
上述代码可能会在一个任务的构造器结束之前开始执行另一个任务，这意味着该任务能够访问处于不稳定状态的对象，这是优先选用Executor而不是显式的创建Thread对象的另一个原因。

2.10 加入一个线程

一个线程可以在其他线程之上调用join()方法，其效果是等待一段时间知道第二个线程结束才继续执行。
如果某个线程在另一个线程t上调用t.join()，此线程将被挂起，知道目标线程t结束才恢复(即t.isAlive()返回为假)
支持在join()加上超时参数

在t中调用interrupt()可以中断

class Sleeper extends Thread {
        private int duration;
        public Sleeper(String name,int sleepTime) {
            super(name);
            duration = sleepTime;
            start();
        }

        @Override
        public void run() {
            try {
                sleep(duration);
            } catch (Exception e) {
                System.out.println(getName() + " was interrupted. " + "isInterrupted():" + isInterrupted());
                return;
            }
        }
    }

    class Joiner extends Thread{
        private Sleeper sleeper;
        public Joiner (String name,Sleeper sleeper) {
            super(name);
            this.sleeper = sleeper;
            start();
        }

        @Override
        public void run() {
            try {
                sleeper.join();
            } catch (InterruptedException e) {
                System.out.println("Interrupted");
            }
            System.out.println(getName() + " join completed");
        }
    }

    public class Joining {
        public static void main(String[] args) {
            Sleeper
                    sleeper = new Sleeper("Sleepy",),
                    grumpy = new Sleeper("grumpy",);
            Joiner
                    joiner = new Joiner("Dopey",sleeper),
                    doc = new Joiner("Doc",grumpy);
            grumpy.interrupt();
        }
    }

多线程简单应用–创建有响应的用户界面

```

    class UnResponseUI {
        private volatile double d = 1;
        public UnResponseUI() throws IOException {
            while (d > 0) {
                d = d + (Math.PI + Math.E) / d;
            }
            System.in.read();
        }
    }
    public class ResponseUI extends Thread {
        private static volatile double d = 1;
        public ResponseUI() {
            setDaemon(true);
            start();
        }

        @Override
        public void run() {
            while (true) {
                d = d + (Math.PI + Math.E) / d;
            }
        }

        public static void main(String[] args) throws IOException {
            new ResponseUI();
            System.in.read();
            System.out.println(d);
        }
    }

```

输入的时候等待的时间不同，产生的结果不一样（因为后台进程Daemon在运行，d还在进行计算）

2.11

继续错误的代价由别人来承担，而承认错误的代价由自己承担。

2.12 捕获异常

使用try-catch不能捕获从线程中逃逸的异常。
使用 Thread.UncaughtExceptionHandler.uncaughtException() 会在线程，因为未捕获的异常而临近死亡时被调用。使用方法是在要捕获异常的Thread对象上附着一个 Thread.UncaughtExceptionHandler 。

thread.setUncaughtExceptionHandler(new MyUncaughtExceptionHandler())

```java

    class ExceptionThread2 implements Runnable{

        @Override
        public void run() {
            Thread thread = Thread.currentThread();
            System.out.println("run() by " + thread);
            System.out.println("eh = " + thread.getUncaughtExceptionHandler());
            throw new RuntimeException();
        }
    }

    class MyUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler{

        @Override
        public void uncaughtException(Thread t, Throwable e) {
            System.out.println("caught " + e);
        }
    }

    class HandlerThreadFactory implements ThreadFactory {

    //    生产带有MyUncaughtExceptionHandler的Thread
        @Override
        public Thread newThread(Runnable r) {
            System.out.println(this + " creating new Thread");
            Thread thread = new Thread(r);
            System.out.println("created " + thread);
            thread.setUncaughtExceptionHandler(new MyUncaughtExceptionHandler());
            System.out.println("eh = " + thread.getUncaughtExceptionHandler());
            return thread;
        }
    }

    public class CaptureUncaughtException {
        public static void main(String[] args) {
            ExecutorService executorService = Executors.newCachedThreadPool(new HandlerThreadFactory());
            executorService.execute(new ExceptionThread2());
        }
    }


```

另有 Thread.setDefaultUncaughtExceptionHandler(new MyUncaughtExceptionHandler()) 设置默认的 未捕获异常处理器 。

3 共享受限资源

none

3.1 不正确地访问资源

boolean类型的变量是原子性，意味着诸如赋值和返回值这样的简单操作在发生时没有中断的可能，所以看不到这个域处于执行这些简单操作的过程中的中间状态。

3.2 解决共享资源竞争

当资源被一个任务使用时，在其上加锁。–序列化访问共享资源
这意味着在给定时刻只允许一个任务访问共享资源。通常在代码前面加上一条锁语句来实现的，这就是的在一段时间内只有一个任务可以运行这段代码。
关键字 synchronized 的形式，为防止资源冲突提供内置支持。
要控制对共享资源的访问，得先把它包装到一个对象里，然后把所有要访问这个资源的方法标记为synchronized。
注意：在使用并发时，将域设置为private是非常重要的，否则，synchronized关键字就不能防止其他任务直接访问域，这样就会产生冲突。
一个任务可以多次获得对象的锁。如果一个方法在同一个对象上调用了第二个方法，后者又调用了同一对象的另一个方法，JVM负责跟踪对象被加锁的次数。显然，只有首先获得了锁的任务才能允许继续获取多个锁。
每个访问临界共享资源的方法都必须被同步，佛则他们就不会正确的工作。

使用显式的Lock对象

```java

    public int next() {
        lock.lock();
        try {
            //一系列操作
        } finally {
            lock.unlock();
        }
    }

```

使用 lock() 和 unlock() 管理临界资源，这样做的优势是方便还有就是可以管理异常。
ReentrantLock允许你尝试着获取锁 tryLock() ,这样的好处是，try一下，如果不行，接着去离开执行其他一些事情，而不是等待。

3.3 原子性与易变性

原子性可以应用于除了long和double之外的所有基本类型之上的“简单操作”。对他们的操作是不可分（原子）的来操作内存。
- 因为JVM将64位（long&double）的读取和写入当做两个分离的32位操作。从而发生了上下文切换。
但是如果使用volatile关键字，就会获得原子性。
volatile还保证了这个域的可视性：对该域产生了写操作，那么所有读操作就都可以看到这个修改。这是因为volatile域会立即被写入到主存中，而读取操作就发生在主存中。
如果一个域完全由synchronized方法或语句块来防护，那就不必将其设置为是volatile的
原子操作
- 赋值
- 返回

3.4 原子类

AtomicInteger、AtomicLong、AtomicReference等特殊的原子性变量类，可以消除synchronized关键字。
依赖于锁更安全些。

3.5 临界区

一个方法中的一段代码被称为临界区。

- 也是使用synchronized关键字

```java
    synchronized(syncObject) {
        //This code can be accessed
        //by only one task at a time
    }
```

- 进入此临界区之前必须得到syncObjec对象的锁。如果得不到，就需要等到锁被释放以后，才能进入临界区。

- 设计模式——模板方法：某个类的一些功能在基类中实现，并且其一个或多个抽象方法在派生类中定义（使用abstract类）。

- 宁愿使用同步控制块而不是对整个方法进行同步控制的典型原因：是的其他线程能更多的访问（在安全的情况下尽可能多）

3.6 在其他对象上同步

synchronized块必须给定一个在其上进行同步的对象，并且最合理的方式就是，使用其方法正在被调用的当前对象：synchronized(this)，一旦获取到synchronized锁，那么该对象其他的synchronized方法和临界区就不能被调用了。

class DualSynch {
    private Object object = new Object();
    public synchronized void f() {
        for (int i  = 0;i < 10000; i++) {
            System.out.println("f()");
            Thread.yield();
        }
    }

    public void g() {
        synchronized (object) {
            for (int i =0;i < 10000;i++) {
                System.out.println("g()");
                Thread.yield();
            }
        }
    }
}

public class SyncObject {
    public static void main(String[] args) {
        final DualSynch dualSynch = new DualSynch();
        new Thread() {
            public void run() {
                dualSynch.f();
            }
        }.start();
        dualSynch.g();
    }
}

3.7 线程本地存储

ThreadLocal对象通常当做静态域存储。在创建ThreadLocal时，你只能通过get()和set()方法来访问该对象的内容， private static ThreadLocal<Interger> value = new ThreadLocal<Integer>(){private Random rand = new Random(47);protected synchronized Integer initialValue() {return rand,nextInt(10000);}};
每个单独的线程都被分配了自己的存储，每个都需要跟踪自己的计数值。

4 终结任务

4.1 装饰性花园

```java

    class Count {
        private int count = 0;
        private Random random = new Random(47);
        public synchronized int increment() {
            int temp = count;
            if (random.nextBoolean()) {
                Thread.yield();
            }
            return (count = ++temp);
        }
        public synchronized int value() {
            return count;
        }
    }

    class Entrance implements Runnable {

        private static Count count = new Count();
        private static List<Entrance> entranceList =
                new ArrayList<>();
        private int number = 0;
        private final int id;
        private static volatile boolean canceled = false;

        public static void cancel() {
            canceled = true;
        }

        public Entrance(int id) {
            this.id = id;
            entranceList.add(this);
        }

        @Override
        public void run() {
            while (!canceled) {
                synchronized (this) {
                    ++number;
                }
                System.out.println(this + " Total: " + count.increment());
                try {
                    TimeUnit.MILLISECONDS.sleep(100);
                } catch (InterruptedException e) {
                    System.out.println("sleep interrupted");
                }
            }
            System.out.println("Stopping " + this);
        }

        public synchronized int getValue() {
            return number;
        }

        @Override
        public String toString() {
            return "Entrance " + id + ": " + getValue();
        }

        public static int getTotalCount() {
            return count.value();
        }

        public static int sumEntrances() {
            int sum = 0;
            for (Entrance entrance:
                    entranceList) {
                sum += entrance.getValue();
            }
            return sum;
        }
    }

    public class OrnamentalGarden {
        public static void main(String[] args) throws InterruptedException {
            ExecutorService executorService = Executors.newCachedThreadPool();
            for (int i = 0; i < 5; i++) {
                executorService.execute(new Entrance(i));
            }
            TimeUnit.SECONDS.sleep(3);
            Entrance.cancel();
            executorService.shutdown();
            if (!executorService.awaitTermination(250,TimeUnit.MICROSECONDS)) {
                System.out.println("Some tasks were not terminated!");
            }
            System.out.println("Total: " + Entrance.getTotalCount());
            System.out.println("Sum of Entrances: " + Entrance.sumEntrances());
        }
    }

```

- 这里使用了单个的Count对象，来跟踪花园参观者的数量（将Count当做一个静态域储存）。

- Count.increment()和Count.value()都是synchronized的，用来控制对count域的访问。

- 每个Entrance任务都维护着一个本地值，它包含通过某个特定入口进入的参观者的数量。

- 因为Entrance.canceled是一个volatile布尔标志，而它只会被读取和赋值（不会与其他域组合在一起被读取），所以不需要同步对其的访问就可以安全地操作它。

- 在3秒钟之后，main()向Entrance发送static cancel()消息，然后调用executorService对象的shutdown()，之后调用executorService.awaitTermination()方法，它会等待每个任务结束，如果所有任务在超时时间达到之前全部结束，则返回true，否则返回false。但这并不意味着Entrance对象是无效的，所以依然可以操作List。

4.2 在阻塞时终结

线程状态：
- 新建：被创建时短暂的处于此状态，已经分配了必须的系统资源，并执行了初始化。已经具有获得CPU时间的资格，之后其运行或阻塞由调度器负责。
- 就绪：在这种状态下，只要调度器把时间片分配给线程，线程就可以运行。
- 阻塞：线程能够运行，但有某个条件阻止它的运行。调度器将会忽略这种线程，不会分配给其任何CPU时间。
- 死亡：这种状态的线程是不可调度的。他的任务已经结束，任务死亡的方式通常是从run返回，但是任务的线程还可以被中断。
进入阻塞状态的原因：
- sleep()
- wait()把该线程挂起。直到线程得到了notify()，notifyAll()消息，线程才会进入就绪状态。
- 等待输入输出的完成
- 拿不到对象锁
如果对于处于阻塞状态的任务，你不能等待其到达代码中可以检查其状态值的某一点，因而决定让他主动地终止，那么你就必须强制这个任务跳出阻塞状态。

4.3 中断

如果使用executor的submit()就会持有该任务的上下文。
submit将会返回一个Future

4.4 检查中断

惯用法

```java

    class NeedsCleanUp {
        private final int id;

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

        public void cleanup() {
            System.out.println("Cleaning up:" + id);
        }
    }
    class Blocked3 implements Runnable {
        private volatile double aDouble = 0.0;
        @Override
        public void run() {
            while (!Thread.interrupted()) {
                NeedsCleanUp needsCleanUp = new NeedsCleanUp(1);
                try {
                    System.out.println("Sleeping");
                    TimeUnit.SECONDS.sleep(1);
                    NeedsCleanUp needsCleanUp1 = new NeedsCleanUp(2);
                    try {
                        System.out.println("Calculating");
                        for (int i = 1; i < 2500000; i++) {
                            aDouble = aDouble + (Math.PI + Math.E) / aDouble;
                            System.out.println("Finished time-consuming operation");
                        }
                    } finally {
                        needsCleanUp1.cleanup();
                    }
                } catch (InterruptedException e) {
                    System.out.println("");
                } finally {
                    needsCleanUp.cleanup();
                }
            }
            System.out.println("Exiting via while() test");
        } 
    }
    public class InterruptingIdiom {
        public static void main(String[] args) throws InterruptedException {
            if (args.length != 1) {
                System.out.println("usage java InterruptingIdiom delay--in-mS");
                System.exit(1);
            }
            Thread thread = new Thread(new Blocked3());
            thread.start();
            TimeUnit.MILLISECONDS.sleep(new Integer(args[0]));
            thread.interrupt();
        }
    }

```

5 线程之间的协作

将你使用的线程同时运行多个任务的时候，可以通过锁(互斥)来同步两个任务的行为（一个任务等另一个有锁的任务用完锁）。
使用互斥可以实现任何时刻只有一个任务可以访问。
在一堆任务中，某些可以并行执行，但是某些步骤需要所有的任务都结束之后才能开动。
线程间的握手（协作），这种握手可以通过Object的方法wait()和notify()来安全地实现。

5.1 wait()与notifyAll(）

wait()是使你可以等待某个条件发生变化，而改变这个条件超出了当前方法的控制能力。
通常这个条件将由另一个任务来改变。
忙等待：任务测试这个条件的同时，不断地进行空循环。
wait()不会进行愚蠢的忙等待，她在等待外界条件的变化的时候会将任务挂起，并且只有notify(）或者notifyAll()，这时候任务被唤醒并且检查条件产生变化了没有
wait会释放锁（因为他是挂起任务）；sleep,yield不会释放锁。
当线程wait后，他将释放锁，这非常重要，因为在这之后，其他任务可能获得锁并发生改变，这种改变正是正是使被挂起的任务重新唤醒所感兴趣的变化。
调用wait就意味着：
我已经刚刚做完能做的所有事情，因此我要在这里等待。但是我希望其他的synchronized操作在条件合适的情况下能够执行。
经常的用法wait(不加参数)，即无限期的等下去，知道notify，notifyAll消息。
这三个方法是基类Object的一部分，而不是Thread的一部分。因为这些方法操作的锁也是所有对象的一部分。

一般这些方法都只在同步代码块中使用。

```java

    class Car {
        private boolean waxOn = false;
        public synchronized void waxed() {
            waxOn = true;
            notifyAll();
        }
        public synchronized void buffed() {
            waxOn = false;
            notifyAll();
        }
        public synchronized void waitForWaxing() throws InterruptedException {
            while (waxOn == false) {
                wait();
            }
        }
        public synchronized void waitForBuffed() throws InterruptedException {
            while (waxOn == true) {
                wait();
            }
        }
    }

    class WaxOn implements Runnable {
        private Car car;

        public WaxOn(Car car) {
            this.car = car;
        }

        @Override
        public void run() {
            try {
                while (!Thread.interrupted()) {
                    System.out.println("Wax On!");
                    TimeUnit.MILLISECONDS.sleep(200);
                    car.waxed();
                    car.waitForBuffed();
                }
            } catch (InterruptedException e) {
                System.out.println("Exiting via interrupt");
            }
            System.out.println("Ending Wax On task");
        }
    }
    class WaxOff implements Runnable {
        private Car car;

        public WaxOff(Car car) {
            this.car = car;
        }

        @Override
        public void run() {
            try {
                while (!Thread.interrupted()) {
                    car.waitForWaxing();
                    System.out.println("Wax Off! ");
                    TimeUnit.MILLISECONDS.sleep(200);
                    car.buffed();//任务被挂起并释放当前对象的锁
                }
            } catch (InterruptedException e) {
                System.out.println("Exiting via interrupt! ");
            }
            System.out.println("Ending Wax Off task");
        }
    }

    public class WaxOMatic {
        public static void main(String[] args) throws InterruptedException {
            Car car = new Car();
            ExecutorService executorService = Executors.newCachedThreadPool();
            executorService.execute(new WaxOff(car));
            executorService.execute(new WaxOn(car));
            TimeUnit.SECONDS.sleep(5);
            executorService.shutdownNow();
        }
    }

```

waxed调用notifyAll，这将唤醒在对wait的调用中被挂起的任务。为了使该任务从wait中唤醒，他必须首先重新获得当他进入wait时释放的锁。在这个锁变得可用之前，这个任务是不会被唤醒的。
这个示例强调你必须用一个检查感兴趣条件的while循环包围wait

本质就是检查所感兴趣的特定条件，并在条件不满足的情况下返回到wait，惯用法就是使用while来编写这种代码。

```java

    //        T1
            synchronized (sharedMonitor) {
                <setup condition for T2>;
                sharedMonitor.notify();
            }
    //        T2
            while (someCondition) {
    //            Point1
                synchronized (sharedMonitor) {
                    sharedMonitor.wait();
                }
            }

```

这段代码可能发生错失信号的问题：假设T2对someCondition求值发现其为true，在Point1，线程调度器可能已经切换到了T1，而T1将执行其代码，并执行notify，当T2得以执行时，已经错过了notify，并进入wait，造成死锁。

修改方案：防止在someCondition变量上产生竞争条件

```java
    //        T2
            synchronized (sharedMonitor) {
                while (someCondition) {
                    sharedMonitor.wait();
                }
            }


```

当T1首先执行，当控制返回T2时，他将会发现条件发生了变化，从而不会进入wait，如果T2先执行，那他将进入wait，并且稍后会由T1唤醒。

5.2 notify与notifyAll

可能会有多个任务在单个Car对象上处于wait状态，因此调用notifyAll比notify更安全。
但是当只有一个任务实际处于wait状态，就可以使用notify了。
这是一种优化。
使用notify的时候，在众多等待同一个锁的任务中只有一个会被唤醒，因此如果你希望使用notify，就必须保证被唤醒的是恰当的任务。
为了使用notify：
- 所有任务必须等待相同的条件，因为如果你有多个任务在等待不同的条件，那么你就不会知道是否唤醒了恰当的任务。
- 当条件发生变化时，必须只有一个任务能够从中受益。
- 这是限制对所有可能存在的子类都必须总是起作用的。
如果一条不满足，就必须使用notifyAll。

notifyAll是当因为某个特定锁而被调用时，只有等待这个锁的任务才会被唤醒。

```java

    class Blocker {
        synchronized void waitingCalling() {
            try {
                while (!Thread.interrupted()) {
                    wait();
                    System.out.println(Thread.currentThread() + " ");
                }
            } catch (InterruptedException e) {
    //            OK to exit this way
            }
        }
        synchronized void prod() {notify();}
        synchronized void prodAll() {notifyAll();}
    }
    class Task implements Runnable {
        static Blocker blocker = new Blocker();

        @Override
        public void run() {
            blocker.waitingCalling();
        }
    }
    class Task2 implements Runnable {
        static Blocker blocker = new Blocker();
        @Override
        public void run() {
            blocker.waitingCalling();
        }
    }
    public class NotifyVsNotifyAll {
        public static void main(String[] args) throws InterruptedException {
            ExecutorService executorService =
                    Executors.newCachedThreadPool();
            for (int i = 0; i < 5; i++) {
                executorService.execute(new Task());
            }
            executorService.execute(new Task2());
            Timer timer = new Timer();
            timer.scheduleAtFixedRate(new TimerTask() {
                boolean prod = true;
                @Override
                public void run() {
                    if (prod) {
                        System.out.println("\nnotify() ");
                        Task.blocker.prod();//不会影响Task2的唤醒
                        prod = false;
                    } else {
                        System.out.println("\nnotifyAll() ");
                        Task.blocker.prodAll();//不会影响Task2的唤醒
                        prod = true;
                    }
                }
            },400,400);//run every .4 second
            TimeUnit.SECONDS.sleep(5);
            timer.cancel();
            System.out.println("\nTimer canceled");
            TimeUnit.MILLISECONDS.sleep(500);
            System.out.println("\nShutting down");
            executorService.shutdownNow();
        }
    }

```

本例中，Task和Task2都有自己的Blocker对象，因此每个Task对象都会在Task.blocker上阻塞。而每个Task2都会在Task2.blocker上阻塞。Timer对象被设置为4/10秒执行run方法，交替地在Task.blocker上调用notify和notifyAll。因为前期没有运行过Task2.blocker，所以Task2永远都是被wait的挂起的。
另外，没有另外。

5.3 生产者与消费者

```java

    class Meal {
        private final int orderNum;

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

        @Override
        public String toString() {
            return "Meal " + orderNum;
        }
    }
    class WaitPerson implements Runnable {//服务员
        private Restaurant restaurant;

        public WaitPerson(Restaurant restaurant) {
            this.restaurant = restaurant;
        }

        @Override
        public void run() {
            try {
                while (!Thread.interrupted()) {
                    synchronized (this) {
                        while (restaurant.meal == null) {
                            wait();//...for the chef to 做这个食物
                        }
                    }
                    System.out.println("-Wait person got " + restaurant.meal);
                    synchronized (restaurant.chef) {
                        restaurant.meal = null;
                        restaurant.chef.notifyAll();// Ready for another
                    }
                }
            } catch (InterruptedException e) {
                System.out.println("WaitPerson interrupted");
            }
        }
    }
    class Chef implements Runnable {
        private Restaurant restaurant;
        private int count = 0;

        public Chef(Restaurant restaurant) {
            this.restaurant = restaurant;
        }

        @Override
        public void run() {
            try {
                while (!Thread.interrupted()) {
                    synchronized (this) {
                        while (restaurant.meal != null) {
                            wait();//...for the meal to 被端走
                        }
                    }
                    if (++count == 10) {
                        System.out.println("Out of food, closing");
                        restaurant.exec.shutdownNow();
                    }
                    System.out.println("Order up ! ");
                    synchronized (restaurant.waitPerson) {
                        restaurant.meal = new Meal(count);
                        restaurant.waitPerson.notifyAll();
                        System.out.println("-");
                    }
                    TimeUnit.MILLISECONDS.sleep(100);
                }
            } catch (InterruptedException e) {
                System.out.println("Chef interrupted");
            }
        }
    }
    public class Restaurant {
        Meal meal;
        ExecutorService exec = Executors.newCachedThreadPool();
        WaitPerson waitPerson = new WaitPerson(this);
        Chef chef = new Chef(this);

        public Restaurant() {
            exec.execute(chef);
            exec.execute(waitPerson);
        }

        public static void main(String[] args) {
            new Restaurant();
        }
    }

```

- Chef厨师和waitPerson服务员，之间是生产者与消费者的关系。Restaurant是两者的焦点。

- 在run中，服务员进入wait模式，停止其任务，直至被Chef的notifyAll唤醒。

- 一旦Chef送上Meal并通知WaitPerson，这个Chef就将等待，直至WaitPerson收集到订单并通知Chef，之后Chef就可以做下一份订单了。

- 有一个问题：在并发应用中，某个其他的任务可能会在WaitPerson被唤醒时，会突然插足（调度器调度问题）并拿走订单，唯一安全的方式是使用下面这种wait惯用法（当然要在恰当的同步内部，并采用防止错失信号可能性的程序设计）：

```java

        while(conditionIsNotMet(不满足的条件//意味着满足了条件就不等待了))
            wait();

    ```

*上述厨师与服务生示例代码中错失信号问题处理方式是正确的 *

重要：

- 这种方式可以保证你退出等待循环之前，条件将得到满足，并且如果你收到了关于某事物的通知，而他与这个条件毫无关系，或者在你完全退出等待条件的之前，这个条件发生了变化【错失信号问题】，都可以确保你可以重返等待状态。

- 通过把整个run放到try中，可以使得这两个run都被设计为可以有序地关闭，catch紧挨着run方法的结束括号之前结束，因此，如果这个任务收到了InterruptedException异常，他将在捕获异常之后立即结束。

重要2：

- shutdownNow将向所有由ExecutorService启动的任务发送interrupt（）

- 在典型的生产者-消费者视线中，应使用先进先出队列来存储被生产和消费的对象。

使用显式的Lock和Condition对象

使用互斥并允许任务挂起的基本类是Condition，可以调用其await来挂起一个任务，当外部条件变化，可以使用signal来通知这个任务，从而唤醒一个任务，也可以使用signalAll唤醒所有在这个Condition上被其自身挂起的任务（比notifyAll更安全）。

```java

    class Car {
        private Lock lock = new ReentrantLock();
        private Condition condition = lock.newCondition();
        private boolean waxOn = false;
        public void waxed() {
            lock.lock();
            try {
                waxOn = true;
                condition.signalAll();
            } finally {
                lock.unlock();
            }
        }
        public void buffed() {
            lock.lock();
            try {
                waxOn = false;
                condition.signalAll();
            } finally {
                lock.unlock();
            }
        }
        public void waitForWaxing() throws InterruptedException {
            lock.lock();
            try {
                while (waxOn == false) {
                    condition.await();
                }
            } finally {
                lock.unlock();
            }
        }
        public void waitForBuffing() throws InterruptedException {
            lock.lock();
            try {
                while (waxOn == true) {
                    condition.await();
                }
            } finally {
                lock.unlock();
            }
        }
    }
    class WaxOn implements Runnable {
        private Car car;

        public WaxOn(Car car) {
            this.car = car;
        }

        @Override
        public void run() {
            try {
                while (!Thread.interrupted()) {
                    System.out.println("WaxOn! ");
                    TimeUnit.MILLISECONDS.sleep(200);
                    car.waxed();
                    car.waitForBuffing();
                }
            } catch (InterruptedException e) {
                System.out.println("Exiting via interrupt");
            }
            System.out.println("Ending Wax On task");
        }
    }
    class WaxOff implements Runnable {
        private Car car;

        public WaxOff(Car car) {
            this.car = car;
        }

        @Override
        public void run() {
            try {
                while (!Thread.interrupted()) {
                    car.waitForWaxing();
                    System.out.println("Wax Off! ");
                    TimeUnit.MILLISECONDS.sleep(200);
                    car.buffed();
                }
            } catch (InterruptedException e) {
                System.out.println("Exiting via interrupt");
            }
            System.out.println("Ending Wax Off task");
        }
    }
    public class WaxOMatic2 {
        public static void main(String[] args) throws InterruptedException {
            Car car = new Car();
            ExecutorService executorService = Executors.newCachedThreadPool();
            executorService.execute(new WaxOff(car));
            executorService.execute(new WaxOn(car));
            TimeUnit.SECONDS.sleep(5);//让他工作5秒，因为WaxOff和WaxOn中的run都有while(!Thread.interrupted){}
            executorService.shutdownNow();
        }
    }

```

在Car的构造器中，单个的Lock将产生一个Condition对象，这个对象被用来管理任务间的通信。
每个队lock()的调用都必须紧跟一个try-finally子句，用来保证在所有情况下都可以释放锁。

5.4 生产者-消费者队列

wait和notifyAll以一种非常低级的方式解决了任务互操作问题，即每次交互时都握手。在许多情况下，你应该瞄向更高的抽象级别，即使用同步队列来解决任务协作问题，同步队列在任何时刻都只允许一个任务插入或移除元素。有java.util.concurrent.BlockingQueue接口中提供了这个队列，常用的有LinkedBlockingQueue，ArrayBlockingQueue。

优秀示例吐司：

class Toast {
                public enum Status {DRY, BUTTERED, JAMMED}
                private Status status = Status.DRY;
                private final int id;

                public Toast(int id) {
                    this.id = id;
                }
                public void butter() {
                    status = Status.BUTTERED;
                }
                public void jam() {
                    status = Status.JAMMED;
                }
                public Status getStatus() {
                    return status;
                }
                public int getId() {
                    return id;
                }

                @Override
                public String toString() {
                    return "Toast{" +
                            "status=" + status +
                            ", id=" + id +
                            '}';
                }
            }
            class ToastQueue extends LinkedBlockingDeque<Toast>{}
            class Toaster implements Runnable {
                private ToastQueue toastQueue;
                private int count = ;
                private Random random = new Random();

                public Toaster(ToastQueue toastQueue) {
                    this.toastQueue = toastQueue;
                }

                @Override
                public void run() {
                    try {
                        while (!Thread.interrupted()) {
                            TimeUnit.MILLISECONDS.sleep( + random.nextInt());
            //                Make toast
                            Toast toast = new Toast(count++);
                            System.out.println(toast);
            //                Insert into queue
                            toastQueue.put(toast);
                        }
                    } catch (InterruptedException e) {
                        System.out.println("Toaster interrupted! ");
                    }
                    System.out.println("Toaster off");
                }
            }
            //apply butter to toast
            class Butterer implements Runnable {
                private ToastQueue dryQueue,butteredQueue;

                public Butterer(ToastQueue dryQueue, ToastQueue butteredQueue) {
                    this.dryQueue = dryQueue;
                    this.butteredQueue = butteredQueue;
                }

                @Override
                public void run() {
                    try {
                        while (!Thread.interrupted()) {
            //                Blocks until next toast is available
                            Toast toast = dryQueue.take();
                            toast.butter();
                            System.out.println(toast);
                            butteredQueue.put(toast);
                        }
                    } catch (InterruptedException e) {
                        System.out.println("Butterer interrupted");
                    }
                    System.out.println("Butterer off");
                }
            }
            //apply jam to buttered toast
            class Jammer implements Runnable {
                private ToastQueue butteredQueue,finishedQueue;

                public Jammer(ToastQueue butteredQueue, ToastQueue finishedQueue) {
                    this.butteredQueue = butteredQueue;
                    this.finishedQueue = finishedQueue;
                }

                @Override
                public void run() {
                    try {
                        while (!Thread.interrupted()) {
            //                blocks until next piece of toast is available;
                            Toast toast = butteredQueue.take();
                            toast.jam();
                            System.out.println(toast);
                            finishedQueue.put(toast);
                        }
                    } catch (InterruptedException e) {
                        System.out.println("Jammer interrupted");
                    }
                    System.out.println("Jammer off");
                }
            }
            //Consume the toast
            class Eater implements Runnable {
                private ToastQueue finishedQueue;
                private int counter = ;

                public Eater(ToastQueue finishedQueue) {
                    this.finishedQueue = finishedQueue;
                }

                @Override
                public void run() {
                    try {
                        while (!Thread.interrupted()) {
            //                Blocks until next toast is available;
                            Toast toast = finishedQueue.take();
                            if (toast.getId() != counter++ || toast.getStatus() != Toast.Status.JAMMED) {
                                System.out.println(">>>>>> Error: " + toast);
                                System.exit();
                            } else {
                                System.out.println("Chomp! " + toast);
                            }
                        }
                    } catch (InterruptedException e) {
                        System.out.println("Eater interrupted");
                    }
                    System.out.println("Eater off");
                }
            }
            public class ToastOMatic {
                public static void main(String[] args) throws InterruptedException {
                    ToastQueue dryQueue = new ToastQueue(),
                            butteredQueue = new ToastQueue(),
                            finishedQueue = new ToastQueue();
                    ExecutorService executorService =
                            Executors.newCachedThreadPool();
                    executorService.execute(new Toaster(dryQueue));
                    executorService.execute(new Butterer(dryQueue,butteredQueue));
                    executorService.execute(new Jammer(butteredQueue,finishedQueue));
                    executorService.execute(new Eater(finishedQueue));
                    TimeUnit.SECONDS.sleep();
                    executorService.shutdownNow();
                }
            }

一台机器具有三个任务：制作吐司，给吐司抹黄油，给抹好黄油的吐司抹果酱。最有有人来吃。
这个示例没有任何显式的同步（即使用Lock或者synchronized的同步），因为同步由队列（其内部是同步的）和系统的设计隐式地管理——每片toast在任何时刻都只有一个任务在操作。
因为队列的阻塞，使得处理过程将被自动的挂起和恢复。
BlockingQueue的简化十分明显。显式地调用wait和notifyAll时存在的类和类之间的耦合被消除了，因为每个类都只和它的BlockingQueue通信。

重要： BlockingQueue的几个方法的api

摘自多线程之BlockingQueue中 take、offer、put、add的一些比较
- 《Java编程思想》——并发读书笔记《Java编程思想》——并发读书笔记2 基本的线程机制3 共享受限资源4 终结任务5 线程之间的协作6 死锁7 新类库中的构件8 仿真9 性能调优10 活动对象11 总结
- offer:
  - 将指定元素插入此队列中（如果立即可行且不会违反容量限制），成功时返回 true，如果当前没有可用的空间，则返回 false，不会抛异常：
- put:
  - 将指定元素插入此队列中，将等待可用的空间.通俗点说就是>maxSize 时候，阻塞，直到能够有空间插入元素
- take:
  - 获取并移除此队列的头部，在元素变得可用之前一直等待。queue的长度 == 0 的时候，一直阻塞
- add:
  - 和collection的add一样，没什么可以说的。如果当前没有可用的空间，则抛出 IllegalStateException。

5.5 任务间使用管道进行输入/输出

提供线程功能的类库就以“管道”的形式对线程间的输入输出提供支持。它们在Java输入输出类库中的对应物就是PipedWriter类（允许任务向管道写）和PipedReader类（允许不同任务从一个管道中读取）。这个模型可以看成是“生产者-消费者”问题的变体，这里的管道就是一个封装好的解决方案。

```java

    class Sender implements Runnable {
        private Random random = new Random(47);
        private PipedWriter writer = new PipedWriter();

        public PipedWriter getWriter() {
            return writer;
        }

        @Override
        public void run() {
            try {
                while (true) {
                    for (char c = 'A'; c <= 'z'; c++) {
                        writer.write(c);
                        TimeUnit.MILLISECONDS.sleep(random.nextInt(500));
                    }
                }
            } catch (InterruptedException e) {
                System.out.println("writer got InterruptedException !");
            } catch (IOException e) {
                System.out.println("writer got IOException !");
            }
        }
    }
    class Receiver implements Runnable {
        private PipedReader reader;

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

        @Override
        public void run() {
            try {
                while (true) {
    //                blocks until characters are there:
                    System.out.println("Read: " + (char)reader.read() + ".");
                }
            } catch (IOException e) {
                System.out.println("Receiver got IOException !");
            }
        }
    }
    public class PipedIO {
        public static void main(String[] args) throws IOException, InterruptedException {
            Sender sender = new Sender();
            Receiver receiver = new Receiver(sender);
            ExecutorService executorService = Executors.newCachedThreadPool();
            executorService.execute(sender);
            executorService.execute(receiver);
            TimeUnit.SECONDS.sleep(10);
            executorService.shutdownNow();
        }
    }


```

可以看到PipedReader与最普通的IO之间最重要的差异是PipedReader是可以中断的。

6 死锁

某个任务在等待另一个任务，而后者有等待别的任务，这样一直下去，直到这个链条上的任务有在等待第一个任务释放锁。这得到了一个任务之间相互等待的连续循环，没有哪个线程能继续。这被称之为死锁。

以下四个条件同时满足时，就会发生死锁：
- 互斥条件。人物使用的资源中至少有一个是不能共享的。这里的一根Chopstick一次就只能被一个Philosopher使用。
- 至少有一个任务他必须持有一个资源且正在等待获取一个当前被别的任务持有的资源。也就是说，要发生死锁，Philosopher必须拿着一根，等另一根。
- 资源不能被任务抢占，任务必须把资源释放当做普通事件。Philosopher很有礼貌，他们不会从其他Philosopher那里抢Chopstick。
- 必须有循环等待，这时，一个任务等待其他任务所持有的资源。后者又在等待另一个任务所持有的资源，这样一直下去，直到有一个任务在等待第一个任务所持有的资源，使得大家都被锁住。破坏此条件最容易。
如果最后一个Philosopher被初始化成先拿左边的，后拿右边的。那么这个Philosopher永远不会阻止其右边的Philosopher拿起他们的一对筷子。

7 新类库中的构件

7.1 CountDownLatch

它被用来同步一个或者多个任务，强制他们等待由其他任务执行的一组操作完成。

```java

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

        public TaskPortion(CountDownLatch latch) {
            this.latch = latch;
        }

        @Override
        public void run() {
            try {
                doWork();
                latch.countDown();
            } catch (InterruptedException e) {
    //            Accepted this way to exit
            }
        }

        public void doWork() throws InterruptedException {
            TimeUnit.MILLISECONDS.sleep(random.nextInt(2000));
            System.out.println(this + "completed");
        }

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

        public WaitingTask(CountDownLatch latch) {
            this.latch = latch;
        }

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

        @Override
        public String toString() {
            return String.format("WaitingTsk %1$-3d ", id);
        }
    }
    public class CountDownLatchDemo {
        static final int SIZE = 100;

        public static void main(String[] args) {
            ExecutorService executorService = Executors.newCachedThreadPool();
    //        All must share a single CountDownLatch object:
            CountDownLatch latch = new CountDownLatch(SIZE);
            for (int i = 0; i < 10; i++) {
                executorService.execute(new WaitingTask(latch));
            }
            for (int i = 0; i < SIZE; i++) {
                executorService.execute(new TaskPortion(latch));
            }
            System.out.println("Launched all tasks");
            executorService.shutdown();
        }
    }


```

- TaskPortion将随机休眠一段时间，以模仿这部分工作完成[portion:部分]，而WaitingTask表示系统中必须等待的部分，他要等待到问题的初始部分完成为止。所有任务都使用了在main()中定义的同一个单一的CountDownLatch。

7.2 CyclicBarrier

其适用于这种场景：你希望创建一组任务，他们并行地执行任务，然后在进行下一个步骤之前等待，直至所有任务都完成（看起来有些像join（））。它使得所有的并行任务都将在栅栏处列队，因此可以一致地向前移动。

class Horse implements Runnable {
    private static int counter = ;
    private final int id = counter++;
    private int strides = ;
    private static Random random = new Random();
    private static CyclicBarrier cyclicBarrier;

    public Horse(CyclicBarrier c) {
        cyclicBarrier = c;
    }

    public synchronized int getStrides() {
        return strides;
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                synchronized (this) {
                    strides += random.nextInt();
                }
                cyclicBarrier.await();
            }
        } catch (InterruptedException e) {
//            A legitimate way to exit
        } catch (BrokenBarrierException e) {
//            This one we want to know about
            throw new RuntimeException(e);
        }
    }

    @Override
    public String toString() {
        return "Horse{" +
                "id=" + id +
                " ";
    }
    public String tracks() {
        StringBuilder builder = new StringBuilder();
        for (int i = ; i < getStrides(); i++) {
            builder.append("*");
        }
        builder.append(id);
        return builder.toString();
    }
}
public class HorseRace {
    static final int FINISH_LINE = ;
    private List<Horse> horseList = new ArrayList<>();
    private ExecutorService executorService = Executors.newCachedThreadPool();
    private CyclicBarrier cyclicBarrier;

    public HorseRace(int nHorses,final int pause) {
        cyclicBarrier = new CyclicBarrier(nHorses, new Runnable() {
            @Override
            public void run() {
                StringBuilder builder = new StringBuilder();
                for (int i = ; i < FINISH_LINE; i++) {
                    builder.append("=");
                }
                System.out.println(builder);
                for (Horse horse:
                        horseList) {
                    System.out.println(horse.tracks());
                }
                for (Horse horse:
                        horseList) {
                    if (horse.getStrides() >= FINISH_LINE) {
                        System.out.println(horse + "won!");
                        executorService.shutdownNow();
                        return;
                    }
                    try {
                        TimeUnit.MILLISECONDS.sleep(pause);
                    } catch (InterruptedException e) {
                        System.out.println("barrier-action sleep interrupted");
                    }
                }
            }
        });
        for (int i = ; i < nHorses; i++) {
            Horse horse = new Horse(cyclicBarrier);
            horseList.add(horse);
            executorService.execute(horse);
        }
    }

    public static void main(String[] args) {
        int nHorse = ;
        int pause = ;
        if (args.length > ) {
            int n = new Integer(args[]);
            nHorse = n >  ? n :nHorse;
        }
        if (args.length > ) {
            int p = new Integer(args[]);
            pause = p > - ? p:pause;
        }
        new HorseRace(nHorse,pause);
    }
}

可以向CyclicBarrier提供一个“栅栏动作”，他是一个Runnable，当计数值到达0时自动执行——这是CyclicBarrier和CountDownLatch之间的另一个区别。这里，栅栏动作是作为匿名内部类创建的，他被提交给了CyclicBarrier的构造器。
CyclicBarrier使得每匹马都要执行为了向前移动所必须执行的所有工作，然后必须在栅栏处等待其它所有的马都准备完毕。当所有的马，都向前移动时，CyclicBarrier将自动调用Runnable栅栏动作任务，按顺序现实马和终点线的位置。
一旦所有的任务都越过了栅栏，他就会自动的为下一回合比赛做好准备。

7.3 DelayQueue

这是一个无界的BlockingQueue，用于放置实现了Delayed接口的对象，其中的对象只能在其到期时才能从队列中取走。这种队列是有序的，即对头对象的延迟到期的时间最长。如果没有任何延迟到期，那么就不会有任何头元素，并且poll（）将返回null（正因为这样，你将不能把null放进这种队列中）。
下面的示例，其中的Delayed对象自身就是任务，而DelayedTaskConsumer将最“紧急”的任务（到期时间最长的任务）从队列中取出，然后运行它。
注意：这样DelayQueue就成为了优先级队列的一种变体。

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() + NANOSECONDS.convert(delta,MILLISECONDS);
        sequence.add(this);//这种用法需要注意！！
    }

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

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

    @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 executorService;
        public EndSentinel(int delay,ExecutorService executorService){
            super(delay);
            this.executorService = executorService;
        }
        public void run() {
            for (DelayedTask task:
                    sequence) {
                System.out.println(task.summary() + " ");
            }
            System.out.println();
            System.out.println(this + "Calling shutdownNow()");
            executorService.shutdownNow();
        }
    }
}
class DelayTaskConsumer implements Runnable {

    private DelayQueue<DelayedTask> taskDelayQueue;

    public DelayTaskConsumer(DelayQueue<DelayedTask> taskDelayQueue) {
        this.taskDelayQueue = taskDelayQueue;
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                taskDelayQueue.take().run();//Run task with th current thread
            }
        } catch (InterruptedException e) {
//            Acceptable way t exit
        }
        System.out.println("Finished DelayedTaskConsumer");
    }
}
public class DelayQueueDemo {
    public static void main(String[] args) {
        Random random = new Random();
        ExecutorService executorService = Executors.newCachedThreadPool();
        DelayQueue<DelayedTask> taskDelayQueue = new DelayQueue<>();
//        Fill with tasks that have random delays;
        for (int i = ; i <; i++) {
            taskDelayQueue.put(new DelayedTask(random.nextInt()));
        }
//        Set the stopping point
        taskDelayQueue.add(new DelayedTask.EndSentinel(,executorService));
        executorService.execute(new DelayTaskConsumer(taskDelayQueue));
    }
}

DelayedTask包含了一个称为sequence的List，它保存了任务被创建的顺序，因此我们可以看到顺序是按照实际发生的顺序执行的。
Delayed接口的方法getDelay（），它可以用来告诉延迟到期有多长时间，或者延迟在多长时间之前已经到期。
System.nanoTime()产生的时间是以纳秒为单位的。你可以转换delta的值，方法是声明他的单位以及你希望以什么单位来表示。
为了排序，Delayed接口还继承了Comparable接口，因此必须实现compareTo()。
嵌套的EndSentinel类提供了一种关闭所有食物的途径，具体做法是将其放置到队列的最后一个元素。
因为DelayedTaskConsumer自身是一个任务，所以他有自己的Thread，他可以使用这个线程来运行从队列中获取的所有任务。由于任务是按照队列的优先级的顺序执行的，因此在本例中不需要启动任何单独的线程来运行DelayedTask。
从输出中可以看到，任务创建的顺序对执行顺序没有任何影响，任务是按照所期望的延迟顺序执行的。

7.4 PriorityBlockingQueue

class PrioritizedTask implements Runnable,Comparable<PrioritizedTask> {

    private Random random = new Random();
    private static int counter = ;
    private final int id = counter++;
    private final int priority;
    protected static List<PrioritizedTask> sequence =
            new ArrayList<>();

    public PrioritizedTask(int priority) {
        this.priority = priority;
        sequence.add(this);
    }

    @Override
    public int compareTo(PrioritizedTask o) {
        return priority < o.priority ?  :
                (priority > o.priority ? - : );
    }

    @Override
    public void run() {
        try {
            TimeUnit.MILLISECONDS.sleep(random.nextInt());
        } catch (InterruptedException e) {
//            Acceptable way to exit;
        }
        System.out.println(this);
    }

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

    public String summary() {
        return "(" + id + ":" + priority + ")";
    }
    public static class EndSentinel extends PrioritizedTask {
        private ExecutorService executorService;
        public EndSentinel(ExecutorService executorService) {
            super(-);
            this.executorService = executorService;
        }
        public void run() {
            int count = ;
            for (PrioritizedTask prioritizedTask:
                    sequence) {
                if (++count %  == ) {
                    System.out.println();
                }
            }
            System.out.println();
            System.out.println(this + " Calling shutdownNow()");
            executorService.shutdownNow();
        }
    }
}
class PrioritizedTaskProducer implements Runnable {

    private Random random = new Random();
    private Queue<Runnable> queue;
    private ExecutorService executorService;

    public PrioritizedTaskProducer(Queue<Runnable> queue, ExecutorService executorService) {
        this.queue = queue;
        this.executorService = executorService;//Used for EndSentinel
    }

    @Override
    public void run() {
//        Unbounded queue; never blocks
//        FIll it up fast with random priorities;
        for (int i = ; i <; i++) {
            queue.add(new PrioritizedTask(random.nextInt()));
            Thread.yield();
        }
//        Trickle in highest-priority jobs;
        try {
            for (int i = ; i < ; i++) {
                TimeUnit.MILLISECONDS.sleep();
                queue.add(new PrioritizedTask());
            }
//            Add jobs,lowest priority first;
            for (int i = ; i < ; i++) {
                TimeUnit.MILLISECONDS.sleep();
            }
//                A sentinel to stop all the tasks
            queue.add(new PrioritizedTask.EndSentinel(executorService));
        } catch (InterruptedException e) {
//            Acceptable way to exit;
        }
        System.out.println("Finished PrioritizedTaskProducer");
    }
}
class PrioritizedTaskConsumer implements Runnable {
    private PriorityBlockingQueue<Runnable> queue;
    public PrioritizedTaskConsumer(PriorityBlockingQueue<Runnable> queue) {
        this.queue = queue;
    }
    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
//                Use current thread to run the task;
                queue.take().run();
            }
        } catch (InterruptedException e) {
//            Acceptable way to exit;
        }
        System.out.println("Finished PrioritizedTaskConsumer");
    }
}
public class PriorityBlockingQueueDemo {
    public static void main(String[] args) {
        Random random = new Random();
        ExecutorService executorService = Executors.newCachedThreadPool();
        PriorityBlockingQueue<Runnable> queue =
                new PriorityBlockingQueue<>();
        executorService.execute(new PrioritizedTaskProducer(queue,executorService));
        executorService.execute(new PrioritizedTaskConsumer(queue));
    }
}

这是一个很基础的优先级队列，它具有可阻塞的读取操作。上面是一个示例，其中在优先级队列中的对象是按照优先级顺序从队列中出现的任务。PrioritizedTask被赋予了一个优先级数字，以此来提供这种顺序。
与前一个示例相同，PrioritizedTask对量的创建序列被记录在sequenceList中，用于和实际的执行顺序相比较。run（）方法将休眠一小段随机时间，然后打印对象信息，而EndSentinel提供了和前面相同的功能，要确保他是队列中最后一个对象。
PrioritizedTaskProducer和PrioritizedTaskConsumer通过PriorityBlockingQueue彼此连接。因为这种队列的阻塞性提供了所有必需的同步，所以你应该注意到了，这里不需要任何显式的同步——不必考虑当你从这种队列中读取时，其中是否有元素，因为这个队列在没有元素时，将直接阻塞读取者。

7.5 使用ScheduledExecutor的温室控制器

通过使用schedule（）（运行一次任务）或者scheduleAtFixedRate（）（每隔规则的时间重复执行任务），你可以将Runnable设置为在将来的某个时刻执行。

public class GreenhouseScheduler {
    private volatile boolean light = false;
    private volatile boolean water = false;
    private String thermostat = "Day";
    public synchronized String getThermostat() {
        return thermostat;
    }
    public synchronized void setThermostat(String value) {
        thermostat = value;
    }
    ScheduledThreadPoolExecutor executor =
            new ScheduledThreadPoolExecutor();
    public void schedule(Runnable event, long delay) {
        executor.schedule(event,delay, TimeUnit.MILLISECONDS);
    }
    public void
    repeat(Runnable event, long initialDelay, long period) {
        executor.scheduleAtFixedRate(event,initialDelay,period,TimeUnit.MILLISECONDS);
    }
    class LightOn implements Runnable {
        @Override
        public void run() {
//            put hardware control code here to
//            physically turn on the light
            System.out.println("Turning on lights");
            light = true;
        }
    }
    class LightOff implements Runnable {
        @Override
        public void run() {
            System.out.println("Turning off lights");
            light = false;
        }
    }
    class WaterOn implements Runnable {
        @Override
        public void run() {
            System.out.println("Turning greenhouse water on");
            water = true;
        }
    }
    class WaterOff implements Runnable {
        @Override
        public void run() {
            System.out.println("Turning greenhouse water off");
            water = false;
        }
    }
    class ThermostatNight implements Runnable {
        @Override
        public void run() {
            System.out.println("Thermostat to night setting");
            setThermostat("Night");
        }
    }
    class ThermostatDay implements Runnable {
        @Override
        public void run() {
            System.out.println("Thermostat to day setting");
            setThermostat("Day");
        }
    }
    class Bell implements Runnable {
        @Override
        public void run() {
            System.out.println("Bing!");
        }
    }
    class Terminate implements Runnable {
        @Override
        public void run() {
            System.out.println("Terminating");
            executor.shutdownNow();
//            Must start a separate task to do this job;
//            since the scheduler has been shut down;
            new Thread() {
                public void run() {
                    for (DataPoint dataPoint:
                            data) {
                        System.out.println(dataPoint);
                    }
                }
            }.start();
        }
    }
//    New feature: data collection
    static class DataPoint {
        final Calendar time;
        final float temperature;
        final float humidity;
        public DataPoint(Calendar calendar, float temp, float hum) {
            time = calendar;
            temperature = temp;
            humidity = hum;
        }
        public String toString() {
            return time.getTime() +
                    String.format(
                    " temperature: %1$.1f humidity: %2$.2f",
            temperature,humidity);
        }
    }
    private Calendar lastTime = Calendar.getInstance();
    {//Adjust date t the half hour
        lastTime.set(Calendar.MINUTE,);
        lastTime.set(Calendar.SECOND,);
    }
    private float lastTemp = f;
    private int tempDirection = +;
    private float lastHumidity = f;
    private int humidityDirection = +;
    private Random random = new Random();
    List<DataPoint> data = Collections.synchronizedList(new ArrayList<DataPoint>());
    class CollectData implements Runnable {
        @Override
        public void run() {
            System.out.println("Collecting data");
            synchronized (GreenhouseScheduler.this) {
//                Pretend the interval is longer than it is;
                lastTime.set(Calendar.MINUTE,lastTime.get(Calendar.MINUTE) + );
//                One in 5 chances of reversing the direction;
                if (random.nextInt() == ) {
                    tempDirection = -tempDirection;
                }
//                Store previous value
                lastTemp = lastTemp +
                        tempDirection * random.nextFloat();
                if (random.nextInt() == ) {
                    humidityDirection = -humidityDirection;
                }
                lastHumidity = lastHumidity + humidityDirection * random.nextFloat();
//                Calendar must be cloned, otherwise all
//                DataPoints hold references to the same lastTime
//                For a basic object like Calendar, clone() is OK
                data.add(new DataPoint((Calendar)lastTime.clone(),
                        lastTemp, lastHumidity));
            }
        }
    }

    public static void main(String[] args) {
        GreenhouseScheduler greenhouseScheduler = new GreenhouseScheduler();
        greenhouseScheduler.schedule(greenhouseScheduler.new Terminate(),);
//        Former "Restart" class not necessary
        greenhouseScheduler.repeat(greenhouseScheduler.new Bell(),,);
        greenhouseScheduler.repeat(greenhouseScheduler.new ThermostatNight(),,);
        greenhouseScheduler.repeat(greenhouseScheduler.new LightOn(),,);
        greenhouseScheduler.repeat(greenhouseScheduler.new LightOff(),,);
        greenhouseScheduler.repeat(greenhouseScheduler.new WaterOn(),,);
        greenhouseScheduler.repeat(greenhouseScheduler.new WaterOff(),,);
        greenhouseScheduler.repeat(greenhouseScheduler.new ThermostatDay(),,);
        greenhouseScheduler.repeat(greenhouseScheduler.new CollectData(),,);
    }
}

收集温室内的温度和湿度读数。DataPoint可以持有并显示单个的数据段，而CollectData是被调度的任务，它在每次运行时，都可以产生仿真数据，并将其添加到Greenhouse的List中。
注意：volatile和synchronized在适当的场合都得到了应用，以防止任务之间的互相干涉。在持有DataPoint的List中的所有方法都是synchronized的，这是因为在List被创建时，使用了java.util.Collection实用工具synchronizedList()。

7.6 Semaphore

正常的锁（来自concurrent.locks或内建的synchronized锁）在任何时候只允许一个任务访问一项资源，而 计数信号量 允许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);
//        Load pool with objects that can be checked out;
        for (int i = ; i < size; i++) {
            try {
//                Assumes a default constructor
                items.add(classObject.newInstance());
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    }
    public T checkOut() throws InterruptedException {
        available.acquire();//如果没有可用的，available将会在这儿阻塞
        return getItem();
    }
    public void checkIn(T x) {
        if (releaseItem(x)) {
            available.release();
        }
    }
    public synchronized T getItem() {
        for (int i = ; i < size; i++) {
            if (!checkedOut[i]) {
                checkedOut[i] = true;
                return items.get(i);
            }
        }
        return null;
    }
    private synchronized boolean releaseItem(T item) {
        int index = items.indexOf(item);
        if (index == -) return false;// Not in the list
        if (checkedOut[index]) {
            checkedOut[index] = false;
            return true;
        }
        return false;
    }
}

在这个简化的形式中，构造器使用newInstance()来把对象加载到池中。如果你需要一个新对象，那么可以调用checkOut()，并且在使用完之后，将其递交给checkIn。
boolean类型的数组checkedOut可以跟踪被签出的对象，并且可以通过getItem（）和releaseItem（）方法来管理。而这些都将由Semaphore类型的available来加以确保，因此，在checkOut（）中，如果没有任何信号量许可证可用（意味着在池中没有更多的对象了），available将阻塞调用过程。在checkIn（）中，如果被签入的对象有效，则会向信号量返回一个许可证。
创建一个Fat类，该类创建代价高昂。

public class Fat {
    private volatile double d;
    private static int counter = ;
    private final int id = counter++;
    public Fat() {
        for (int i = ; i < ; i++) {
            d += (Math.PI + Math.E) / (double)i;
        }
    }
    public void operation() {
        System.out.println(this);
    }

    @Override
    public String toString() {
        return "Fat{" +
                "d=" + d +
                ", id=" + id +
                '}';
    }
}

我们在池中管理这些对象，以限制这个构造器所造成的代价高昂的影响。接下来创建一个任务，他将签出Fat对象，持有一段时间之后再将它们签入。以此来测试Pool类。

class CheckoutTask<T> implements Runnable {
    private static int counter = ;
    private final int id = counter++;
    private Pool<T> pool;

    public CheckoutTask(Pool<T> pool) {
        this.pool = pool;
    }

    @Override
    public void run() {
        try {
            T item = pool.checkOut();
            System.out.println(this + "checking out " + item);
            TimeUnit.SECONDS.sleep();
            System.out.println(this + "checking in " + item);
            pool.checkIn(item);
        } catch (InterruptedException e) {
//            Acceptable way to terminate
        }
    }

    @Override
    public String toString() {
        return "CheckoutTask{" +
                "id=" + id +
                ", pool=" + pool +
                '}';
    }
}
public class SemaphoreDemo {
    final static int SIZE = ;

    public static void main(String[] args) throws Exception {
        final Pool<Fat> pool =
                new Pool<>(Fat.class,SIZE);
        ExecutorService executorService =
                Executors.newCachedThreadPool();
        for (int i = ; i < SIZE; i++) {
            executorService.execute(new CheckoutTask<>(pool));
        }
        System.out.println("All CheckoutTasks created");
        List<Fat> fatList = new ArrayList<>();
        for (int i = ; i < SIZE; i++) {
            Fat fat = pool.checkOut();
            System.out.println(i + ": main() thread checked out ");
            fat.operation();
            fatList.add(fat);
        }
        Future<?> blocked = executorService.submit(new Runnable() {
            @Override
            public void run() {
                try {
//                Semaphore prevents additional checkout
//                so call is blocked
                    pool.checkOut();
                } catch (InterruptedException e) {
                    System.out.println("checkOut() Interrupted");
                }

            }
        });
        TimeUnit.SECONDS.sleep();
        blocked.cancel(true);
        System.out.println("Checking in objects in " + fatList);
        for (Fat f :
             fatList) {
           pool.checkIn(f);
        }
        for (Fat f :
                fatList) {
            pool.checkIn(f);
        }
        executorService.shutdown();
    }
}

在main中，创建了一个持有Fat对象的Pool，而一组ChekoutTask则开始操作这个Pool，然后，main线程签出池中的Fat对象，但是并不签入他们。一旦池中所有的对象都被签出，Semaphore将不再允许执行任何签出操作。blocked的run方法因此会被阻塞，2s之后，cancel方法被调用，以此来挣脱Future的束缚。注意：冗余的签入将会被Pool忽略。

semaphore.acquire()，获取信号量，没信号量可用时，将进行阻塞

semaphore.release(); 释放信号量
本例中Semaphore作用对象是Pool，最多SIZE个任务访问这个pool资源。

7.7 Exchanger

Exchanger是两个任务之间交换对象的栅栏。当这些任务进入栅栏时，他们各自拥有一个对象，当他们离开时，他们都拥有之前有对象持有的对象。
典型的应用场景：一个任务在创建对象，这些对象的生产代价高昂，而另一个任务在消费这些对象。通过这种方式，可以有更多的对象在被创建的同时被消费。
示例：我们将创建生产者与消费者任务，它们经由泛型和Generator，可以工作于任何类型的对象，然后我们将他们应用于Fat类。ExchangerProducer和ExchangerConsumer使用一个List 作为要交换的对象，他们都包含一个用于这个List的Exchanger。当你调用Exchanger.exchanger()方法时，他将阻塞直至对方任务调用它自己的exchange()方法，那时，这两个exchange()方法将全部完成，而List则被互换。

class ExchangerProducer<T> implements Runnable {
    private Surrogate.Generator generator;
    private Exchanger<List<T>> exchanger;
    private List<T> holder;

    public ExchangerProducer(Surrogate.Generator generator, Exchanger<List<T>> exchanger, List<T> holder) {
        this.generator = generator;
        this.exchanger = exchanger;
        this.holder = holder;
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                for (int i = ; i < ExchangerDemo.size; i++) {
                    holder.add(generator.next());
//                    Exchange full for empty
                    holder = exchanger.exchange(holder);
                }
            }
        } catch (InterruptedException e) {
//            OK to terminate this way
        }
    }
}
class ExchangerConsumer<T> implements Runnable {
    private Exchanger<List<T>> exchanger;
    private List<T> holder;
    private volatile T value;

    public ExchangerConsumer(Exchanger<List<T>> exchanger, List<T> holder) {
        this.exchanger = exchanger;
        this.holder = holder;
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                holder = exchanger.exchange(holder);
                for (T x:
                        holder) {
                    value = x;
                    holder.remove(x);
                }
            }
        } catch (InterruptedException e) {
//            OK to terminate this way;
        }
        System.out.println("Final value: " + value);
    }
}
public class ExchangerDemo {
    static int size = ;
    static int delay = ;

    public static void main(String[] args) throws InterruptedException {
        if (args.length > ) {
            size = new Integer(args[]);
        }
        if (args.length > ) {
            delay = new Integer(args[]);
        }
        ExecutorService executorService = Executors.newCachedThreadPool();
        Exchanger<List<Fat>> exchanger = new Exchanger<>();
        List<Fat> 
                producerList = new CopyOnWriteArrayList<>(),
                consumerList = new CopyOnWriteArrayList<>();
        executorService.execute(new ExchangerProducer<Fat>(exchanger,BasicGenerator.create(Fat.class),producerList));
        executorService.execute(new ExchangerConsumer<Fat>(exchanger,consumerList));
        TimeUnit.SECONDS.sleep(delay);
        executorService.shutdownNow();
    }
}

Exchanger类中的exchange(String x) 方法具有阻塞的特点，也就是说此方法被调用后等待其他线程来获取数据，如果没有其他线程取得数据，则就会一直阻塞等待下去。

Exchanger类中的exchange(String x, long timeout, TimeUnit unit)查看官方API可知这个方法的作用是在制定的时间内没有其他线程获取数据，则会抛出异常

Exchanger类详解

8 仿真

9 性能调优

9.1 比较各类互斥技术

测试技巧：
- 意识到在编译过程中和在运行时实际会发生什么
- 在这，测试不同的锁的性能时，我们只有在这些互斥存在竞争的情况下，才能看到真正的性能差异，因此必须有多个任务尝试着访问互斥代码区。
- 在这，编译器看到synchronized关键字时，可能会执行特殊的优化，甚至有可能会注意到这个程序是单线程的。编译器甚至可能会识别出counter被递增的次数是固定数量的，因此会预先计算出其结果。
- 模板方法设计模式：将所有公用代码都放置到基类中，并将所有不同的代码隔离在导出类的accumulate()和read()方法中（这两个方法都是abstract）。
- 为了保险起见，初始测试执行了两次，而第一次的结果被丢弃，因为他包含了初试线程的创建。
使用Lock通常会比使用synchronized要高效的多，而且synchronized的开销变化太大，而Lock相对比较一致。
一个很好的习惯：只互斥那些你绝对必须互斥的部分。
具有实际意义的做法：以synchronized关键字入手，只有在性能调优时才替换为Lock对象这种做法。

以更加传统的互斥方法入手，只有在性能方面的需求能够明确指示时，再替换为Atomic。
以上建议的一个原因：synchronized的阅读性好，（代码被阅读的次数远多于被编写的次数。）

9.2 免锁容器

Java SE5特别添加了新的容器，通过使用更灵巧的技术来消除加锁，从而提高线程安全的性能。

这些免锁容器背后的通用策略是：对容器的修改可以与读取操作同时发生，只要读取者只能看到完成修改的结果即可。修改是在容器数据结构的某个部分的一个单独的副本（有时是整个数据结构的副本）上执行的，并且这个副本在修改过程中是不可视的。只有当修改完成后，被修改的结构才会自动地与主数据结构进行交换，之后读取者就可以看到这个修改了。

- 在CopyOnWriteArrayList中，写入将导致创建整个底层数组的副本，而源数组将保留在原地，使得复制的数组在被修改时，读取操作可以安全地执行。CopyOnWrite容器也是一种读写分离的思想，读和写不同的容器。

- Java并发编程：并发容器之CopyOnWriteArrayList（转载）

- CopyOnWriteArraySet将使用CopyOnWriteArrayList来实现其免锁行为。

- ConcurrentHashMap和ConcurrentLinkedQueue使用了类似的技术，允许并发的读取和写入，但是容器中只有部分内容而不是整个容器可以被复制和修改。然而，任何修改在完成之前，读取者仍旧不能看到它们。And ConcurrentHashMap不会抛出ConcurrentModificationException异常。

乐观锁

只要你主要是从免锁容器中读取，那么它就会比其synchronized对应物快许多，因为获取和释放锁的开销被省掉了。如果需要向免锁容器中执行少量写入，那么情况仍旧如此，但是什么算“少量”，这是一个很有意思的问题。

public abstract class Tester<C> {
    static int testReps = ;
    static int testCycles = ;
    static int containerSize = ;
    abstract C containerInitializer();
    abstract void startReadersAndWriters();
    C testContainer;
    String testId;
    int nReaders;
    int nWriters;
    volatile long readResult = ;
    volatile long readTime = ;
    volatile long writeTime = ;
    CountDownLatch endLatch;
    static ExecutorService executorService =
            Executors.newCachedThreadPool();
    Integer[] writeData;
    Tester(String testId, int nReaders, int nWriters) {
        this.testId = testId + " " +
                nReaders + "r " + nWriters + "w";
        this.nReaders = nReaders;
        this.nWriters = nWriters;
        writeData = Generated.array(Integer.class, new RandomGenerator.Integer(),containerSize);
        for (int i = ; i < testReps; i++) {
            runTest();
            readTime = ;
            writeTime = ;
        }
    }
    void runTest() {
        endLatch = new CountDownLatch(nReaders + nWriters);
        testContainer = containerInitializer();
        startReadersAndWriters();
        try {
            endLatch.await();
        } catch (InterruptedException e) {
            System.out.println("endLatch interrupted");
        }
        System.out.printf("%-27s %14d\n","readTime + writeTime =",readTime + writeTime);
    }
    abstract class TestTask implements Runnable {
        abstract void test();
        abstract void putResults();
        long duration;
        public void run() {
            long startTime = System.nanoTime();
            test();
            duration = System.nanoTime() - startTime;
            synchronized (Tester.this) {
                putResults();
            }
            endLatch.countDown();
        }
    }

    public static void main(String[] args) {
        if (args.length > ) {
            testReps = new Integer(args[]);
        }
        if (args.length > ) {
            testCycles = new Integer(args[]);
        }
        if (args.length > ) {
            containerSize = new Integer(args[]);
        }
        System.out.printf("%-27s %14s %14s\n",
                "Type", "Read time", "Write time");
    }
}

abstract方法containerInitializer()返回将被测试的初始化之后的容器，它被存储在testContainer域中，另一个abstract方法startReadersAndWriters()启动读取者和写入者任务，他们将读取和修改待测容器。不同的测试在运行时具有数量变化的读取者和写入者，这样就可以观察到锁竞争（针对synchronized容器而言）和写入（针对免锁容器而言）的效果。
我们向构造器提供了各种有关测试的信息（参数标识符应该是自解释的），然后他会调用runTest方法repetitions次，runTest将创建一个CountDownLatch（因此测试可以知道所有任务何时完成）、初始化容器，然后调用startReadersAndWriters，并等待他们全部完成。
每个Reader和Writer类都基于TestTask，他可以度量其抽象方法test的执行时间，然后在一个synchronized块中调用putResults去存储结果。
为了使用这个框架（其中你可以识别出模板方法设计模式），我们必须让想要测试的特定类型的容器继承Tester，并提供合适的Reader类和Writer类。

abstract class ListTest extends Tester<List<Integer>> {
    ListTest(String testId, int nReaders, int nWriters) {
        super(testId, nReaders, nWriters);
    }
    class Reader extends TestTask {
        long result = ;
        void test() {
            for (long i = ; i <testCycles; i++) {
                for (int index = ; index < containerSize; index++) {
                    result += testContainer.get(index);
                }
            }
        }
        void putResults() {
            readResult += result;
            readTime += duration;
        }
    }
    class Writer extends TestTask {
        void test() {
            for (long i =; i < testCycles; i++) {
                for (int index = ; index < containerSize; index++) {
                    testContainer.set(index,writeData[index]);
                }
            }
        }
        void putResults() {
            writeTime +=duration;
        }
    }

    @Override
    void startReadersAndWriters() {
        for (int i = ; i < nReaders; i++) {
            executorService.execute(new Reader());
        }
        for (int i = ; i < nWriters; i++) {
            executorService.execute(new Writer());
        }
    }
}
class SynchronizedArrayListTest extends ListTest {
    List<Integer> containerInitializer() {
        return Collections.synchronizedList(
                new ArrayList<Integer>(
                        new CountingIntegerList(containerSize)
                )
        );
    }

     SynchronizedArrayListTest(int nReaders, int nWriters) {
        super("Synched ArrayList",nReaders,nWriters);
    }

}
class CopyOnWriterArrayListTest extends ListTest {
    List<Integer> containerInitializer() {
        return new CopyOnWriteArrayList<Integer>(
                new CountingIntegerList(containerSize)
        );
    }
    CopyOnWriterArrayListTest(int nReaders,int nWriters) {
        super("Synched ArrayList",nReaders,nWriters);
    }
}
public class ListComparisons {
    public static void main(String[] args) {
        Tester.initMain(args);
        new SynchronizedArrayListTest(,);
        new SynchronizedArrayListTest(,);
        new SynchronizedArrayListTest(,);
        new CopyOnWriterArrayListTest(,);
        new CopyOnWriterArrayListTest(,);
        new CopyOnWriterArrayListTest(,);
        Tester.executorService.shutdown();
    }
}

结论：从输出中可以看到，synchronized ArrayList无论读取者和写入者的数量是多少，都具有大致相同的性能——读取者与其他读取者竞争锁的方式与写入者相同。但是，CopyOnWriteArrayList在没有写入者时，速度回快许多，并且在有5个写入者时，速度仍旧明显的快。看起来你应该尽量使用CopyOnWriteArrayList，对列表写入的影响并没有超过短期同步整个列表的影响。
当然，你必须在你的具体应用中尝试这两种不同的方式，以了解到底哪个更好一些。

比较各种Map实现

通过比较synchronizedHashMap和ConcurrentHashMap
- 向ConcurrentHashMap添加写入者的影响甚至还不如CopyOnWriteArrayList明显，这是因为ConcurrentHashMap使用了一种不同的技术，它可以明显地最小化写入所造成的影响。

9.3 乐观加锁

尽管Atomic对象将执行像decrementAndGet这样的原子性操作，但是某些Atomic类还允许你执行所谓的“乐观加锁”。这意味着当你执行某项计算时，实际上没有使用互斥，但是在这项计算完成，并且你准备更新这个Atomic对象时，你需要使用一个称为compareAndSet的方法。在这里，你将新值与旧值一起提交给这个方法，如果旧值与他在Atomic对象中发现的值不一致，那么这个操作就失败——这意味着某个其他的任务已经于此操作执行期间修改了这个对象。
记住，我们在正常情况下将使用互斥（synchronized或者Lock）来防止多个任务同时修改一个对象，但是这里我们是“乐观的”，因为我们保持数据为未锁定状态，并希望没有任何其他任务插入修改它。
所有这些又都是以性能的名义执行的——通过使用Atomic来替代synchronized或Lock，可以获得性能上的好处。
如果compareAndSet操作失败会发生什么？
- 如果compareAndSet操作失败，那么就必须决定做些什么，这是一个非常重要的问题，因为如果不能执行某些恢复操作，那么你就不能使用这项技术，而转去使用传统的互斥。
- 你可能去重试这个操作，如果在第二次成功，那么万事大吉；或者可能会忽略这次失败，直接结束——在某些仿真中，如果数据点丢失，在重要的框架中，这就是最终需要做的事情（当然，你必须很好的了解你的模型，以了解情况是否真的如此）。

public class FastSimulation {
    static final int N_ELEMENTS = ;
    static final int N_GENS = ;
    static final int N_EVOLVERS = ;
    static final AtomicInteger[][] GRID =
            new AtomicInteger[N_ELEMENTS][N_GENS];
    static Random random = new Random();
    static class Evoler implements Runnable {
        @Override
        public void run() {
            while (!Thread.interrupted()) {
//                Randomly select an element to work on
                int element = random.nextInt(N_ELEMENTS);
                for (int i = ; i < N_GENS; i++) {
                    int previous = element - ;
                    if (previous < ) previous = N_ELEMENTS - ;
                    int next = element + ;
                    if (next >= N_ELEMENTS) next = ;
                    int oldValue = GRID[element][i].get();
//                    Perform some kind of modeling calculation
                    int newValue = oldValue +
                            GRID[previous][i].get() + GRID[next][i].get();
                    newValue /= ;   // Average the three values
                    if (!GRID[element][i].compareAndSet(oldValue,newValue)) {
                        /***
                         * Policy here to deal with failure. Here, we
                         * just report it and ignore it; our model
                         * will eventually deal with it,
                         */
                        System.out.println("OldValue changed from " + oldValue);
                    }
                }
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = ; i < N_ELEMENTS; i++) {
            for (int j = ; j < N_GENS; j++) {
                GRID[i][j] = new AtomicInteger(random.nextInt());
            }
        }
        for (int i = ; i < N_EVOLVERS; i ++) {
            executorService.execute(new Evoler());
        }
        TimeUnit.SECONDS.sleep();
        executorService.shutdownNow();
    }
}

所有元素都被置于数组内，这被认为有助于提高性能。每个Evolver对象会用他前一个元素和后一个元素来平均它的值，如果更新时失败，那么将直接打印这个值并继续执行。注意：在这个程序中，没有出现任何的互斥。

9.4 ReadWriteLock

ReadWriteLock对向数据结构相对不频繁地写入，但是有多个任务要经常读取这个数据结构的这类情况进行了优化。ReadWriteLock是的你可以同时有多个读取者，只要他们都不试图写入即可。如果写锁已经被其他任务持有，那么任何读取者都不能访问，直到这个写锁被释放为止。
ReadWriteLock是否能用来提高性能完全取决于数据的读取与修改的频率相比较的结果。最好就是用试验来证明ReadWriteLock对你的程序有没有帮助。

public class ReaderWriterList<T> {
    private ArrayList<T> lockedList;
//    Make the ordering fair;
    private ReentrantReadWriteLock lock =
        new ReentrantReadWriteLock(true);
    public ReaderWriterList(int size, T initialValue) {
        lockedList = new ArrayList<T>(
                Collections.nCopies(size,initialValue)
        );
    }
    public T set(int index, T element) {
        Lock wLock = lock.writeLock();
        wLock.lock();
        try {
            return lockedList.set(index,element);
        } finally {
            wLock.unlock();
        }
    }
    public T get(int index) {
        Lock rLock = lock.readLock();
        rLock.lock();
        try {
//            Show that multiple readers
//            may acquire the read lock
            if (lock.getReadHoldCount() > ) {
                System.out.println(lock.getReadHoldCount());
            }
            return lockedList.get(index);
        } finally {
            rLock.unlock();
        }
    }

    public static void main(String[] args) throws Exception {
        new ReaderWriterListTest(,);
    }
}
class ReaderWriterListTest {
    ExecutorService executorService =
            Executors.newCachedThreadPool();
    private final static int SIZE = ;
    private static Random random = new Random();
    private ReaderWriterList<Integer> list =
            new ReaderWriterList<>(SIZE,);
    private class Writer implements Runnable {
        @Override
        public void run() {
            try {
                for (int i = ; i < ; i++) {
                    list.set(i , random.nextInt());
                    TimeUnit.MILLISECONDS.sleep();
                }
            } catch (InterruptedException e) {
//                    Acceptable way to exit
            }
            System.out.println("Writer finished, shutting down");
            executorService.shutdownNow();
        }
    }
    private class Reader implements Runnable {
        @Override
        public void run() {
            try {
                while (!Thread.interrupted()) {
                    for (int i = ; i < SIZE; i++) {
                        list.get(i);
                        TimeUnit.MILLISECONDS.sleep();
                    }
                }
            } catch (InterruptedException e) {
//                    Acceptable way to exit
            }
        }
    }
    public ReaderWriterListTest(int readers, int writers) {
        for (int i = ; i <readers; i++) {
            executorService.execute(new Reader());
        }
        for (int i = ; i <writers; i++) {
            executorService.execute(new Writer());
        }
    }
}

解释：ReadWriteList可以持有固定数量的任何类型的对象。你必须向构造器提供所希望的列表尺寸和组装这个列表时所用的初始对象。set方法要获取一个写锁，以调用底层的ArrayList.set()，而get方法要获取一个读锁，以调用底层的ArrayList.get()。另外，get（）将检查是否已经有多个读取者获取了读锁，如果是，则将显示这种读取者的数量，以证明可以有多个读取者获得读取锁。
为了测试ReaderWriterLIst，ReaderWriterListTest为ReaderWriterList创建了读取者和写入者，注意，写入者远少于读取者。

10 活动对象

由于必须花很大力气去实现防止多线程任务彼此互相干涉，所以有一种可替换的方式被称为 活动对象 或 行动者 ，之所以称这些对象为“活动的”，是因为每个对象都维护着他自己的工作器线程和消息队列，并且所有对这种对象的请求都将进入队列排队，实现了任何时刻都只能运行其中一个。
因此，有了活动对象，我们就可以串行化消息而不是方法，这意味着不再需要防备一个任务在其循环的中间被中断这种问题了。
当你向一个活动对象发送消息时，这条消息会转变为一个任务，该任务会被插入到这个对象的队列中，等待在以后的某个时刻运行。Java SE5的Future在实现这种模式时会派上用场。它有两个方法，可以将方法调用排进队列。

public class ActiveObjectDemo {
    private ExecutorService executorService =
            Executors.newSingleThreadExecutor();
    private Random random =
            new Random();
//    Insert a random delay to produce the effect
//    of a calculation time
    private void pause(int factor) {
        try {
            TimeUnit.MILLISECONDS.sleep(
                     + random.nextInt()
            );
        } catch (InterruptedException e) {
            System.out.println("sleep() interrupted");
        }
    }
    public Future<Integer>
    calculateInt(final int x, final int y) {
        return executorService.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                System.out.println("starting " + x + " + " + y);
                pause();
                return x + y;
            }
        });
    }
    public Future<Float>
    calculateFloat(final float x, final float y) {
        return executorService.submit(new Callable<Float>() {
            @Override
            public Float call() throws Exception {
                System.out.println("starting " + x + " + " + y);
                return x + y;
            }
        });
    }
    public void shutdown() {
        executorService.shutdown();
    }

    public static void main(String[] args) {
        ActiveObjectDemo demo = new ActiveObjectDemo();
//        Prevents ConcurrentModificationException
        List<Future<?>> results =
                new CopyOnWriteArrayList<Future<?>>();
        for (float f= f; f < f; f += f) {
            results.add(demo.calculateFloat(f,f));
        }
        for (int i = ; i < ; i++) {
            results.add(demo.calculateInt(i,i));
        }
        System.out.println("All asynch calls made");
        while (results.size() > ) {
            for (Future<?> future:
                    results) {
                if (future.isDone()) {
                    try {
                        System.out.println(future.get());
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                    results.remove(future);
                }
            }
        }
        demo.shutdown();
    }
}

由对Excecutores.newSingleThreadExecutor()的调用产生的单线程执行器维护者它自己的无界阻塞队列，并且只有一个线程从该队列中取走任务并执行他们直至完成。
用submit提交一个新的Callable对象，以响应对这些方法的调用，这样就可以把方法调用转变为 消息 ，而每个活动对象方法的返回值都是一个具有泛型参数的Future，而这个泛型参数就是该方法中实际的返回类型。通过这种方式，方法调用几乎可以立即返回，调用者可以使用Future来发现何时任务完成，并收集实际的返回值。
在main中，创建了一个LIst

11 总结

可以运行多个独立任务。
必须考虑当这些任务关闭时，可能出现的所有问题。
任务可能会在共享资源上彼此干涉。互斥（锁）是用来防止这种冲突的基本工具。
如果任务设计的不够仔细，就有可能会发生死锁。
- 明白什么时候应该使用并发，什么时候应该避免使用并发湿肺娼馆间的。使用它的原因主要是：
- 要处理很多任务，他们交织在一起，应用并发能够更有效的使用计算机（包括在多个CPU上透明的分配人物的能力）。
- 要能够更好的组织代码。
- 要更便于用户使用。【经典案例是：在长时间的下载过程中监视“停止”按钮是否被按下。】
- 线程的一个额外好处是他们提供了轻量级的执行上下文切换（大约100条指令），而不是重量级的进程上下文切换（要上千条指令）。因为一个给定进程内的所有线程共享相同的内存空间，轻量级的上下文切换只是改变了程序的执行序列和局部变量。进程切换（重量级的上下文切换）必须改变所有内存空间。
多线程的主要缺陷有：
- 等待共享资源的时候性能降低。
- 需要处理线程的额外CPU话费。
- 糟糕的程序设计导致不必要的复杂度。
- 有可能产生一些病态行为,如饿死、竞争、死锁和活锁（多个运行各自任务的线程使得整体无法完成）。
- 不同平台导致的不一致性。
因为多个线程可能共享一个资源，所以你必须确定多个线程不会同时读取和改变这个资源，这就是线程产生的最大难题。
线程的最大数量，通常依赖于操作系统和JVM；他可以是不足一百个线程，也可能是几千个线程。

《Java编程思想》——并发读书笔记《Java编程思想》——并发读书笔记2 基本的线程机制3 共享受限资源4 终结任务5 线程之间的协作6 死锁7 新类库中的构件8 仿真9 性能调优10 活动对象11 总结