Java多线程编程实战指南读书笔记三

线程数设置的原则

对于CPU密集型线程，考虑到这些线程执行任务值消耗的主要是处理器资源，我们可以将线程数设置为$N_{cpu}$,有事也可以设置为$N_{cpu}+1$
对于I/O密集型线程，考虑到I/O操作可能导致上下文切换，为这样的线程设置过多的线程数会导致过多的额外系统开销，如果一个工作线程能够满足旧不需要设置更多的线程，如果一个工作者线程不够用，我们可以考虑将这类线程的数量设置为$2*N_{cpu}$

等待与通知：wait/notify

一个线程因为执行目标动作所需的保护条件未满足而被暂停的过程就被称为等待，
一个线程更新了系统的状态，使得其它线程所需要的保护条件得以满足的时候唤醒那些被暂停的线程的过程被称为通知。

wait/notify的作用与用法

Java平台中Object.wait()/Object.wait(long)和Object.notify()/Object.notifyAll()可用于实现等待和通知。

使用Object.wait()实现等待

synchronized (someObject){
            while (保护条件不成立){
                //调用wait方法暂停当前线程
                someObject.wait();
            }
            //执行目标动作
            doAction();
}

其中保护条件是一个包含共享变量的布尔表达式.当这些共享变量被其它线程更新之后使相应的保护条件得以成立时，这些线程会通知等待线程。由于一个线程只有在持有一个对象的内部锁的情况下才能够调用该对象的wait方法，所以Object.wait()调用总是放在相应对象所引导的临界区之中。包含上述模板代码的方法被称为受保护方法。受保护方法包含三个要素：保护条件，暂停当前线程和目标动作。

因执行wait方法而被暂停的线程被称为对象上的等待线程。由于同一个对象的同一个方法（wait方法）可以被多个线程执行，因此一个对象可能存在多个等待线程。wait方法会以原子操作的方式使其执行线程（当前线程）暂停，并使该线程释放其持有的对应的内部锁。其它线程在该线程保护条件成立的时候执行相应的notify方法，可以唤醒someOBject上的任意的一个等待线程。被唤醒的等待线程在其占用处理器继续运行的时候，需要再次申请该对象的内部锁。

使用notify实现通知

//通知方法
synchronized (someObject){
            //更新保护条件涉及的共享变量
            updateSharedState();
            //通知等待的线程
            someObject.notify();
}

通知方法包含两个要素：更新共享变量，唤醒其它线程。

notify的执行线程持有的相应的对象的内部锁只有在notify调用所在的临界区代码执行结束后才会被释放，而notify本身并不会将这个内部。为了减少锁的争用，建议尽量把notify放在临界区代码结尾的地方
notify唤醒的线程是相应对象上的任意等待线程。而且等待线程和通知线程是同步在同一个对象之上的两种线程。

wait/notify的开销及问题

过早唤醒。假设一组等待/通知线程同步在对象someObject上，初始状态下均不成立，当线程N1更新了共享变量state1使得N1执行notifyAll方法唤醒所有线程，当其它线程的保护条件可能并不成立，这使得线程唤醒后还得睡眠。
信号丢失。如果等待线程在执行wait前没有先判断保护条件是否已经成立，那么存在这种情形，通知线程在该等待线程进入临界区之前就已经更新了相关的共享变量，使得相应的保护条件成立并进行了通知，但是此时等待线程还没有被暂停，自然就不会唤醒了。也就是说唤醒的信号丢失了。
欺骗性唤醒。等待线程也可能存在没有其它任何线程执行notify和notifyAll线程的情况下被唤醒。
上下文切换问题。wait/notify的时候可能导致较多的上下文切换

Java条件变量

JDK1.5 之后引入的新的标准库类java.util.concurent.locks.Condition接口可以作为wait/notify的替代品。
Lock.newCondition()的返回值就是一个Condition实例。因此调用任何一个显示锁的实例的newCondition方法都可以创建一个相应的Condition接口。Condition.await()/signal()也要求其执行线程持有创建该Condition实例的显示锁。每个Condition实例内部都维护乐一个用于存储等待线程的队列。

假设cond1和cond2是两个不同的Condition实例，一个线程执行Cond1.await()会导致其被暂停，并被存入cond1的等待队列。cond1.signal会使cond1的等待队列中的任意线程被唤醒。调用cond1.signalAll()会使cond1的等待队列中的所有线程被唤醒。而cond2等待队列中的任何一个等待线程都不会受此影响。

private final Lock lock=new ReentrantLock();
    private final Condition condition=lock.newCondition();
    public void aGuaredMethod() throws InterruptedException{
        lock.lock();
        try {
            while (保护条件不成立){
                condition.await();
            }
            
            //执行目标动作
            doAction();
            
        }finally {
            lock.unlock();
        }
    }
    
    
    public void anNotificationMethod() throws InterruptedException{
        lock.lock();
        try {
            //更新共享变量
            changeState();
            
            condition.signal();
        }finally {
            lock.unlock();
        }
    }

使用Condition接口可以很好的解决过早唤醒的问题。它还解决了Objetc.wait(long)方法无法区分其返回是因为等待超时还是被通知的。Conditon.awaitUntil(Date deadline)可以用于实现待超时时间限制的等待，并且该方法的返回值能够区分该方法调用是由于等待超时而返回还是由于其它线程执行了相应的条件变量的signal/signalAll方法。

Condition.await()/signal()执行线程需要持有创建相应条件变量的显示锁。

值得注意的是等待线程被唤醒后再次申请显示锁的这段时间内，其它线程可能已经抢先获得了锁并更新了条件变量使得等待线程所需要的保护条件重新不成立，所以我们在Condition.awaitUntil(Data)返回true（等待未超时）的情况下我们可以选择继续等待。

倒计时协调器 CountDownLatch

Thread.join()可以实现一个线程等待另一个线程结束。有时候一个线程可能只需要等待其它线程执行的特定操作结束即可，而不需要等待该线程终止。我们可以使用条件变量来实现这个，也可以使用倒计时协调器。

ConutDownLatch内部维护了一个用于表示未完成的先决操作数量的计数器。每调用一次countDown（）相应实例的计数器值会减少1（先决条件完成了一个），await()方法相当于一个受保护的方法，其保护条件为计数器的值为0(先决条件全部完成)。
因此，当计数器的值不为0的时候，CountDownLatch.await()的执行线程会被暂停。这些线程就被称为CountDownLatch上的等待线程。CountDownlatch.countDown()相对于一个通知方法，它会在计数器达到0的时候唤醒相应实例上的等待线程。
因为CountDownLatch内部封装了对“全部先决条件的等待与通知的逻辑”，所以客户端代码在调用await,countDown方法都无需加锁。

CountDownLatch内部计数器值达到0后器值就恒定不变，后续执行该CountDownLatch实例await方法的任何一个线程都不会被暂停。为了避免线程永远等待，需要将CountDown调用放在代码无论怎样都能执行到的地方，或者设置等待时间

对于同一个CountDownLatch实例latch，latch.countDown()的执行线程在执行该方法之前所执行的任何内存操作对等待线程在latch.await()调用返回之后的代码是可见的且有序的。

栅栏

有些时候多个线程可能需要相互等待对方执行到代码中的某个地方。这是整些线程才能够继续执行。JDK 1.5开始引入了一个java.util.concurrent.CyclicBarrier，该类可以实现这种等待。使用CyclicBarrier实现等待的线程被称为参与方。参与方只需要执行await()方法就可以实现等待。CyclicBarrier内部维护了一个显示锁，这使其总是可以在参与方中区分出一个最后执行CyclicBarrier.await()的线程。除了最后一个线程外的任何参与方执行await都会导致该线程被暂停。

典型的应用场景

使迭代算法并发化。在并发化的迭代算法中，迭代操作是由多个工作者线程并行执行的。CyclicBarrier可以来实现执行迭代操作的任何一个工作者线程必须等大其它工作者线程完成当前迭代操作的情况下才能继续其下一轮的迭代操作，以便形成迭代操作的中间结果作为下一轮迭代的基础（输入）。
在测试代码中模拟高并发。在编写多线程程序的测试代码时，我们常常需要使用邮箱的工作者线程来模拟高并发操作。这些工作者线程中的任意一个线程在执行其操作前必须等待其它线程也准备就绪，使得这些工作者线程能同一时刻开始其操作。

生成者-消费者模型

在生产者消费者模型中，生产者的主要职责时生产产品，产品可以是数据，也可以是任务。消费者的主要职责是消费生产者所生产的产品。这里的消费包括对产品所代表的数据进行加工处理或者执行产品所代表的任务。
生产者和消费者是并发的运行在各自的线程之中的，这意味着运用生产者-消费者模式可以使程序中原本串行的处理得以并发化。因为线程之间无法直接传递数据，因此生产者消费者之间需要一个用于传递产品的传输通道。该传输通道相当于生产者消费者之间的缓冲区。传输通道通常可以使用一个线程安全的队列来实现。

由于生产者和消费者运行在不同的线程中，因此生产者将产品放入传输通道，消费者在从相应的传输通道中取出产品的过程其实就是生产者线程间对象发布到消费者线程的过程，这种对象发布必须是线程安全的

阻塞队列

jdk1.5引入的结口java.util.concurrent.BlockingQueeu定义了一种线程安全的队列：阻塞队列。阻塞队列按照其存储空间的容量是否受限制来划分，可分为有界队列和无界队列。
一般而言，一个方法或操作如果能够导致其执行线程被暂停（生命周期状态为waiting或者blocked），那么我们就称相应的方法/操作为阻塞方法

几种阻塞队列的对比

ArrayBlockingQueue是有界队列，内部使用一个数组作为存储空间，因为数组的存储空间是预先分配好的所以put,take操作本身并不会增加垃圾回收的负担，缺点是在内部实现put，take时使用的是同一个显示锁，可能会导致锁的高争用，进而导致较多的上下文切换。
LinkedBlockingQUeue既能实现无界队列也能实现有界队列。在创建其实例时可以指定容量。它的优点是其内部在实现put,take操作的时候分别使用了两个显示锁，这降低了锁争用的可能性。其缺点是内部使用一个链表来存储数据可能会导致垃圾回收的负担，
SynchronousQueue可以被看座一种特殊的有界队列。它的内部并不维护用于存储队列元素的存储空间。适用于生产者生产一个产品必须等待消费后才能继续生产的场景。

阻塞队列也支持非阻塞式操作，比如BlockingQueue提供的offer(E) poll()。

对虚拟资源的管理

在jdk1.5中引入了标准类库java.util.concurrent.Sernaphore可以用来实现流量控制。我们把代码中所访问的特定资源或执行特定操作的机会同一看作一种资源，这种资源被称为虚拟资源。Semaphore相当于虚拟资源配额管理器，它可以用来控制同一语句内对虚拟资源的访问次数。

为了对虚拟资源进行流量控制，我们必须使用相应代码只有在获得相应配额的情况下才能够访问这些资源。因此，代码在访问虚拟资源前必须先申请相应的配额，并在资源访问结束后返还相应的配额。Semaphore.acquire()/release()分别用于申请配额和返回配额。Semaphore.acquire()申请成功后会立即返回，如果配额不足则会使执行线程等待。release()会使当前配额值加一，并唤醒Semaphore实例的等待队列种的一个任意等待线程。

acquire和release总是配对使用的。
release的调用总是应该放在一个finally块种。
创建Semaphore实例时如果构造器种的参数permits值为一，那么所创建的Semaphore实例相当于一个互斥锁。
配额本身可被看作程序执行特定操作前所需持有的资源，因此对配额的调度也涉及公平性问题。默认情况下，Semaphore采用的是非公平性调度策略。

管道

java类库种提供了PipedOutPutStrean和PipedInputStream，它们可以用来实现线程间直接输入和输出。从代码的角度来看，一个线程的输出可以作为另外一个线程的输入，而不需要借用文件、数据库、网络连接的其它数据交换中介。

PipedOutPutStrean和PipedInputStream适合在单生产者和单消费者吗模式中使用。

双缓冲区Exchanger

但消费者线程消费一个已填充的缓冲区时，另外一个缓冲区可以由生产者线程进行填充，从而实现了数据生成和与消费的并发。这种缓冲技术被称为双缓冲。
jdk 1.5中引入了java.util.concurrent.Exchanger可以用来实现双缓冲。

简单说就是一个线程在完成一定的事务后想与另一个线程交换数据，则第一个先拿出数据的线程会一直等待第二个线程，直到第二个线程拿着数据到来时才能彼此交换对应数据。其定义为 Exchanger 泛型类型，其中 V 表示可交换的数据类型

Exchanger()：无参构造方法

V exchange(V v)：等待另一个线程到达此交换点（除非当前线程被中断），然后将给定的对象传送给该线程，并接收该线程的对象。

V exchange(V v, long timeout, TimeUnit unit)：等待另一个线程到达此交换点（除非当前线程被中断或超出了指定的等待时间），然后将给定的对象传送给该线程，并接收该线程的对象

当一个线程到达 exchange 调用点时，如果其他线程此前已经调用了此方法，则其他线程会被调度唤醒并与之进行对象交换，然后各自返回；如果其他线程还没到达交换点，则当前线程会被挂起，直至其他线程到达才会完成交换并正常返回，或者当前线程被中断或超时返回

import java.util.concurrent.Exchanger;

public class ThreadStudy08 {

    public static void main(String[] args) throws InterruptedException {
        Exchanger<Integer> exchanger = new Exchanger<Integer>();
        new Producer("", exchanger).start();
        new Consumer("", exchanger).start();
        Utils.randomPauser(1000);
        System.exit(-1);
    }

    static class Producer extends Thread {
        private Exchanger<Integer> exchanger;
        private static int data = 0;
        Producer(String name, Exchanger<Integer> exchanger) {
            super("Producer-" + name);
            this.exchanger = exchanger;
        }

        @Override
        public void run() {
            for (int i=1; i<5; i++) {
                try {
                    Utils.randomPauser(1000);
                    data = i;
                    System.out.println(getName()+" 交换前:" + data);
                    data = exchanger.exchange(data); //等待对方到达交换点
                    System.out.println(getName()+" 交换后:" + data);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }


    static class Consumer extends Thread {
        private Exchanger<Integer> exchanger;
        private static int data = 0;

        Consumer(String name, Exchanger<Integer> exchanger) {
            super("Consumer-" + name);
            this.exchanger = exchanger;
        }

        @Override
        public void run() {
            while (true) {
                data = 0;
                System.out.println(getName() + " 交换前:" + data);
                try {
                    Utils.randomPauser(1000);
                    data = exchanger.exchange(data);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(getName() + " 交换后:" + data);
            }
        }
    }

    }

线程中断机制

Java线程中断机制相当于Java线程与线程间协作的一套协议框架。中断可被看作由一个线程发送给另外一个线程的一种指示，该指示用于表示发起线程希望目标线程停止其正在执行的操作。目标线程可能会满足发起线程的述求，也可能不会 java中每个线程维护一个被称为中断标记的布尔类型变量用于表示相应线程是否收到中断（true为收到）。通过Thread.currentThread().isinterrupted()调用来获取该线程的中断标记值，也可以通过Thread.interrupted()来获取并重置中断标记值，调用一个线程的interrupt()相当于将该线程的中断标记置为true

InterruptedException异常处理及中断响应

能够响应中断的方法通常时在执行阻塞操作前判断中断标志，若中断标志置为true则抛出InterruptedException。

按照惯例，抛出InterruptedException异常的方法，通常会在其抛出该异常时将当前线程中断标记置为false

如果在发起线程给目标线程发送中断的那一刻，目标线程已经由于执行了一些阻塞操作而被暂停，那么Java虚拟机会设置目标线程的中断标记并唤醒线程。

//正确的捕获中断异常的方法
try{
    Thread.sleep(1000)
}catch(InterruptedExceptioin e){
    //保留线程中断标记
    Thread.currentThread().interrupt();
}

应当避免InterruptedExceptioin异常被吞没。

线程停止

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class TerminatableTaskRunner {


    public static void main(String[] args) throws InterruptedException {
        TerminatableTaskRunner runner=new TerminatableTaskRunner(new ArrayBlockingQueue<Runnable>(5));

        runner.submit(()->{
            for(int i=0;i<2;i++){
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                System.out.println("任务A执行一次");
            }

        });
        runner.submit(()->{
            for(int i=0;i<2;i++){
                try {
                  Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                System.out.println("任务B执行一次");
            }
        });

        runner.init();
        Thread.sleep(1000);
        runner.shutdown();
    }
    protected final BlockingQueue<Runnable> channel;

    private volatile Thread workerThread;

    //线程停止标记
    protected volatile boolean inUse =true;

    //待处理的任务计数器
    public final AtomicInteger reservations=new AtomicInteger(0);

    public TerminatableTaskRunner(BlockingQueue<Runnable> channel) {
        this.channel = channel;
        this.workerThread=new workerThread();
    }

    public void init(){
        final Thread t=workerThread;
        if (null != t) {
            t.start();
        }
    }

    public void submit(Runnable task) throws InterruptedException{
        channel.put(task);
        //待执行的任务加一
        reservations.incrementAndGet();
        System.out.println("收到任务");

    }
    public void shutdown(){
        System.out.println("开始停止线程");
        inUse =false;
        final Thread t=workerThread;
        if(t!=null){
            t.interrupt();
        }
    }
    class workerThread extends Thread{
        @Override
        public void run(){
            Runnable task=null;
            try {
                while (true){
                    //线程不再需要，且无待处理的任务
                    if (!inUse &&reservations.get()<=0){
                        break;
                    }
                    //从阻塞队列中拿出一个任务
                    task=channel.take();
                    try {
                        task.run();
                    }catch (Throwable e){
                        e.printStackTrace();
                    }
                    //使得待处理任务数减一
                    reservations.decrementAndGet();
                }
            }catch (InterruptedException e){
                workerThread=null;
            }
            System.out.println("工作线程结束");
        }
    }
}