Java并发学习 & Executor学习 & 异常逃逸 & 同步互斥Best Practice & wait/notify, conditon#await/signal

看了这篇文章：http://www.ciaoshen.com/2016/10/28/tij4-21/ 有一些Java并发的内容，另外查了一些资料。

朴素的Thread

首先，Java中关于线程Thread最基本的事实是：

• 除非通过Native方法将本地线程加入JVM，创建线程唯一的方法就是“创建一个Thread类的实例对象，然后调用它的start()方法。”

其次，关于Thread对象实例的构造，需要注意，start()方法依赖于run()方法：

• 要么传递一个Runnable对象给构造器做参数。
• 要么重写Thread自己的run()方法。

第一种方法是实现Runnable接口。注意，Runnable里面获取线程信息需要用 Thread.currentThread()

package com.company;


class MyRunnable implements Runnable {
    public void run() {
        try {
            Thread.sleep((long)(Math.random() % 5 * 1000 + 1000));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.printf("Here is thread %d
", Thread.currentThread().getId());
    }
}

public class Main {

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

        System.out.println("Hello!");
        MyRunnable myRunnable = new MyRunnable();
        Thread myThread1 = new Thread(myRunnable);
        Thread myThread2 = new Thread(myRunnable);
        myThread1.start();
        myThread2.start();

        // Your Codec object will be instantiated and called as such:
        //System.out.printf("ret:%d
", ret);

        System.out.println();

    }

}

第二种方法是直接继承Thread，需要多继承的，要用上一种Runnable接口的方法。

package com.company;


class MyThread extends Thread {
    public void run() {
        try {
            Thread.sleep((long)(Math.random() % 5 * 1000 + 1000));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.printf("Here is thread %d
", getId());
    }
}

public class Main {

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

        System.out.println("Hello!");
        MyThread myThread1 = new MyThread();
        MyThread myThread2 = new MyThread();
        myThread1.start();
        myThread2.start();

        // Your Codec object will be instantiated and called as such:
        //System.out.printf("ret:%d
", ret);

        System.out.println();

    }

}

Executor和线程池

朴素的Thread对象，对映单个线程。多个Thread对象，多个线程是可以共存的。但会互相竞争资源。Executor创建一个“线程池”的概念，对线程统一管理。

Java SE5的java.util.concurrent包中的执行器（Executor）将为你管理Thread对象，从而简化了并发编程。

实验代码如下（有一些好的注意点）：

package com.company;


import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

class ConcurrentSum {
    private int coreCpuNum;
    private ExecutorService executorService;
    private List<FutureTask<Long>> tasks = new ArrayList<FutureTask<Long>>();

    public ConcurrentSum() {
        coreCpuNum = Runtime.getRuntime().availableProcessors();
        System.out.printf("There's %d cores
", coreCpuNum);
        executorService = Executors.newFixedThreadPool(coreCpuNum);
    }

    class SumCalculator implements Callable<Long> {

        int nums[];
        int start;
        int end;
        public SumCalculator(final int nums[], int start, int end) {
            this.nums = nums;
            this.start = start;
            this.end = end;
        }

        @Override
        public Long call() throws Exception {
            long sum = 0;
            for (int i=start; i<end; i++) {
                sum += nums[i];
            }
            return sum;
        }
    }

    public long sum(int[] nums) {
        int start, end, increment;
        for (int i=0; i<coreCpuNum; i++) {
            // 注意这里分片的方法是非常棒的
            increment = nums.length /  coreCpuNum + 1;
            start = i * increment;
            end = start + increment;
            if (end > nums.length) {
                end = nums.length;
            }
            SumCalculator sumCalculator = new SumCalculator(nums, start, end);
            // FutureTask的构造参数是一个实现了Callable的对象
            FutureTask<Long> task = new FutureTask<Long>(sumCalculator);
            tasks.add(task);
            if (!executorService.isShutdown()) {
                executorService.submit(task);
            }

        }
        return reduce();
    }

    private long reduce() {
        long sum = 0;
        for (int i=0; i<tasks.size(); i++) {
            try {
                sum += tasks.get(i).get();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
        // 如果没有下面这句,那么整个程序不会退出。
        executorService.shutdown();
        return sum;
    }
}

public class Main {

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

        System.out.println("Hello!");

        // main routine
        int arr[] = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
        long sum = new ConcurrentSum().sum(arr);
        System.out.printf("sum: %d
", sum);

        // Your Codec object will be instantiated and called as such:
        //System.out.printf("ret:%d
", ret);

        System.out.println();

    }

}

上面的reduce部分，使用了迭代循环获取各个FutureTask的结果，而如果某个结果还没有返回，则会阻塞。如果希望不阻塞，可以使用CompletionService。

CompletionService对ExecutorService进行了包装，内部维护一个保存Future对象的BlockingQueue。

只有当这个Future对象状态是结束的时候，才会加入到这个Queue中，take()方法其实就是Producer-Consumer中的Consumer。

它会从Queue中取出Future对象，如果Queue是空的，就会阻塞在那里，直到有完成的Future对象加入到Queue中。

所以，先完成的必定先被取出。这样就减少了不必要的等待时间。 

实验代码如下，和上面直接使用ExecutorService有一些区别:

package com.company;


import java.util.concurrent.*;

class ConcurrentSum {
    private int coreCpuNum;
    private ExecutorService executorService;
    private CompletionService<Long> completionService;

    public ConcurrentSum() {
        coreCpuNum = Runtime.getRuntime().availableProcessors();
        System.out.printf("There's %d cores
", coreCpuNum);
        executorService = Executors.newFixedThreadPool(coreCpuNum);
        completionService = new ExecutorCompletionService<Long>(executorService);
    }

    class SumCalculator implements Callable<Long> {

        int nums[];
        int start;
        int end;
        public SumCalculator(final int nums[], int start, int end) {
            this.nums = nums;
            this.start = start;
            this.end = end;
        }

        @Override
        public Long call() throws Exception {
            long sum = 0;
            for (int i=start; i<end; i++) {
                sum += nums[i];
            }
            return sum;
        }
    }

    public long sum(int[] nums) {
        int start, end, increment;
        for (int i=0; i<coreCpuNum; i++) {
            // 注意这里分片的方法是非常棒的
            increment = nums.length /  coreCpuNum + 1;
            start = i * increment;
            end = start + increment;
            if (end > nums.length) {
                end = nums.length;
            }
            SumCalculator sumCalculator = new SumCalculator(nums, start, end);
            // CompletionService直接提交一个实现了Callable的对象
            if (!executorService.isShutdown()) {
                completionService.submit(sumCalculator);
            }

        }
        return reduce();
    }

    private long reduce() {
        long sum = 0;
        for (int i=0; i<coreCpuNum; i++) {
            try {
                sum += completionService.take().get();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
        // 如果没有下面这句,那么整个程序不会退出。
        executorService.shutdown();
        return sum;
    }
}

public class Main {

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

        System.out.println("Hello!");

        // main routine
        int arr[] = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
        long sum = new ConcurrentSum().sum(arr);
        System.out.printf("sum: %d
", sum);

        // Your Codec object will be instantiated and called as such:
        //System.out.printf("ret:%d
", ret);

        System.out.println();

    }

}

经过这样的改动，获取各个线程结果的地方就不会block了。

yield( )让步

和System.gc()方法类似，yield()方法仅仅是“建议”当前线程可以让给其他线程了。但完全不保证会让位。

未捕获异常

异常逃逸：主要原因是抛出异常的线程，和抓异常的代码所在的线程不是一个。这样即使在main函数里面抓异常也是抓不到的。

比如如下代码

package com.company;


import java.util.concurrent.*;

class ConcurrentSum {
    // Runnable
    public class SuperException implements Runnable {

        @Override
        public void run() {
            throw new RuntimeException();
        }
    }
    // Executor
    public void letsGo() {
        ExecutorService executorService = Executors.newCachedThreadPool();
        try {
            executorService.execute(new SuperException());
        } catch (Exception e) {
            System.out.println("Here catch Exception");
            e.printStackTrace();
        } finally {
            System.out.println("Here finally shutdown");
            executorService.shutdown();
        }
    }
}

public class Main {

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

        System.out.println("Hello!");

        // main routine
        new ConcurrentSum().letsGo();

        // Your Codec object will be instantiated and called as such:
        //System.out.printf("ret:%d
", ret);

        System.out.println();

    }

}

就不会抓到线程，命令行输出：

Hello!
Exception in thread "pool-1-thread-1" java.lang.RuntimeException
    at com.company.ConcurrentSum$SuperException.run(Main.java:12)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Here finally shutdown

如上，“异常逃逸”不是说异常就不见了，消失了。其实它还是会冒泡到控制台的。而且自作主张显示在异常报告的第一行。这里的”逃逸”是指异常逃脱了我们try{}catch{}语句对异常的处理。

逃逸的原因很容易猜，因为执行execute()方法的是主线程的Excecutor。
而抛出异常的线程池中被分配来执行run()的某线程。JVM的异常处理是各线程只管自己的事。
所以同理，就算我们把异常处理套到main()方法的主体中也无法捕获异常。因为始终是在主线程里做动作，这是无法处理其他线程里的异常的。

注意，C++里面也是这样。即使是在C++的join调用的外层，包上try-catch也没有用的，还是抓不到异常。

那有没有办法，在主线程里面捕获子线程的异常呢？有的！

重载ThreadFactory里面的newThread方法，在其中加上对UncauhgtExceptionHandler实现类的绑定。如下：

package com.company;


import java.util.concurrent.*;

class ConcurrentSum {
    // Runnable
    class SuperException implements Runnable {

        @Override
        public void run() {
            throw new RuntimeException();
        }
    }

    class MyUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {

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

    class HandlerThreadFactory implements ThreadFactory {

        @Override
        public Thread newThread(Runnable r) {
            System.out.println(this + " createing new Thread");
            Thread t = new Thread(r);
            System.out.println("created " + t);
            t.setUncaughtExceptionHandler(new MyUncaughtExceptionHandler());
            System.out.println("eh = " + t.getUncaughtExceptionHandler());
            return t;
        }
    }

    // Executor
    public void letsGo() {
        ExecutorService executorService = Executors.newCachedThreadPool(new HandlerThreadFactory());
        try {
            executorService.execute(new SuperException());
        } catch (Exception e) {
            System.out.println("Here catch Exception");
            e.printStackTrace();
        } finally {
            System.out.println("Here finally shutdown");
            executorService.shutdown();
        }
    }
}

public class Main {

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

        System.out.println("Hello!");

        // main routine
        new ConcurrentSum().letsGo();

        // Your Codec object will be instantiated and called as such:
        //System.out.printf("ret:%d
", ret);

        System.out.println();

    }

}

独占锁，synchronized关键字

可以加在方法上，如下：

class Mutex implements Runnable{
    private volatile int num=0;    //“private”禁止外部方法调用
    public synchronized void increment(){
    }
}

任何线程如果想要调用increment()方法，必须先获得当前Mutex类实例对象的唯一“独占令牌”，直到increment()方法执行完成，才释放令牌。
在此期间，其他所有希望对同一个Mutex对象执行increment()操作的线程，都必须阻塞等候。

也可以加在代码临界区上：

class Mutex implements Runnable{
    private volatile int num=0;    //“private”禁止外部方法调用
    public void increment(){
        synchronized（this）{
                       ... 
        }
    }
}

synchronized方法里面可以调用本身对象，也可以调用其他对象。

ReentrantLock，乐观锁

除了synchronized之外，另一个选择是使用ReentrantLock，又叫“乐观锁”。用法和效果和synchronized都差不多。差别是它必须显式地创建锁，锁住和解锁。

但ReentrantLock解决资源冲突的机制，和synchronized完全不同。它使用了非阻塞算法（non-blocking algorithms）。简单说就是：乐观地假设操作不会频繁地引起冲突，而是先进行操作，如果没有其他线程争用共享数据，那操作就成功了。如果共享数据被争用，产生了冲突，那就再进行其他的补偿措施（最常见的补偿措施就是不断地重试，直到试成功为止）。

如另一篇博文中所讲（link）：synchronized的加锁机制也是有很多优化的，从偏向锁，到轻量级锁，到重量级锁，逐渐升级。其中，强量级锁比较类似ReentrantLock所采用的CAS机制；偏向锁甚至更加优化，只是在对象某个flag置一个偏向锁的标记以及持有这个偏向锁的ThreadId，然后只要不发生竞争，就没有问题；发生竞争了，就升级到轻量级锁。

非阻塞算法能奏效，基于一个前提条件：需要操作和冲突检测这两个步骤具备原子性，它靠硬件指令来保证，这里用的是 CAS 操作（Compare and Swap）。
进一步研究 ReentrantLock 的源代码，会发现其中比较重要的获得锁的一个方法是 compareAndSetState，这里其实就是调用的 CPU 提供的特殊指令。
直接用单个指令保证原子性。AutomicInteger、AutomicLong、AutomicReference 等特殊的原子性变量类，它们提供的如：
compareAndSet()、incrementAndSet()和getAndIncrement()等方法都使用了 CAS 操作。

看一下AtomicInteger源码是如何保证线程同步的：

public final int getAndSet(int newValue) {
     for (;;) {
         int current = get();
         if (compareAndSet(current, newValue))
             return current;
     }
}
  public final boolean compareAndSet (int expect, int update) {
    return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
}

乐观锁因为没有频繁的上下文切换，效率较高。

关于volatile

• Any write to a volatile variable establishes a happens-before relationship with subsequent reads of that same variable.

这里的“happens-before relationship（偏序关系）”指的就是，必须保证如果值的改变发生在读取之前，那么这个改变要确确实实写进内存，让读取操作“可见”。

粗略说就是：每次值的写入都直接写进内存，而不使用CPU缓存的优化。

线程安全的三个关键词：“互斥性”，“可见性”，“原子性”。

结束线程

ExecutorService#shutdown()：不再接受新任务。

ExecutorService#shutdownNow()：立刻终止所有任务。

ExecutorService#awaitTermination()：阻塞直到所有现有任务完成，然后结束所有线程，关闭线程池。

线程（任务）的中断（interrupt）

Thread#interrupt()方法可以“试图”中断“阻塞中”的线程。注意只能中断处于”阻塞状态“的线程。但：

• sleep阻塞是可以被中断的
• IO阻塞是不可以被中断的
• synchronized阻塞是不可以被中断的

如果任务不是由execute()执行，而是submit()执行，那在返回的Future上调用cancel()，可以有针对性地关闭线程池中的特定任务。

但要强制中断IO阻塞，可以直接关闭底层IO。另外和普通IO不同，nio是可以响应Future的cancel()中断的。

Synchronized的独占锁不可中断，但ReentrantLock的乐观锁是可以中断的。用Reentrant#lockInterruptibly()。因为上面分析过了，乐观锁本质上并没有阻塞冲突线程，它们只是在不断地重试而已。

中断任务的一个惯用法（良好实践）

由于Java的interrupt只能中断“处于阻塞状态中”的任务（虽然对于IO和synchronized锁造成的阻塞也无力中断），所以当线程处于“非阻塞状态”下愉快运行的时候，除非暴力结束线程，看起来我们没有办法中断某个任务。

一个可行的方法是用静态方法Thread.interrupted()判断当前线程是否收到interrupt命令。

try{
    while(!interrupted()){
        //工作代码，一旦收到中断指令，就跳出
    }
}catch(InterruptedException ie){
    //除非是在sleep()状态下被中断，否则不会捕获InterruptedException
}finally{
    //非阻塞状态下被中断后的处理
}

 

wait( ), notify( ), notifyAll( )

wait()阻塞挂起当前线程的同时，释放互斥锁。这点和sleep()不同，sleep()不释放互斥锁。

someObject.notifyAll();
someObject.wait();

先唤醒正在等待某个对象互斥锁的所有线程，然后再阻塞挂起当前线程，释放互斥锁，这样做是安全的。

另外wait()的一个惯用法是：尽量把wait()放在一个while(!condition){wait();}里面。防止醒来后却发现不满足条件的情况。

最后，对某个对象调用wait()和notify(),notifyAll()之前先获得这个对象上的互斥锁。

 

notify( )和notifyAll( )

notify()和notifyAll()的区别在于，notifyAll()唤醒所有排队线程，而notify()只唤醒其中一个线程，但却无法控制唤醒的是哪一个。

notifyAll()的策略就是，在这个锁上等的线程都叫醒。由线程自己判断这次的事务是否和自己有关。

notify()只叫醒一个线程，线程也需要自己判断这次的事务是否和自己有关。但notify()和notifyAll()的区别在于，如果任务和被唤醒的线程无关，继续睡之前，此线程还需要把接力棒传下去唤醒另一个线程，虽然它也不清楚唤醒的是哪个线程。

所以一般来说notifyAll()更合理一些。特殊情况用notify()要小心。

wait( )能被interrupt信号中断

这里有必要再次强调interrupt的有效范围：

• 能中断sleep()阻塞
• 能中断wait()阻塞
• 无法中断synchronized互斥锁阻塞
• 无法中断IO阻塞
• 能中断ReentrantLock的乐观锁（笔者加）

尤其注意，当使用while(!Thread.interrupted())判断时，不要过早拦截InterruptedException导致无法跳出循环。

“生产者-消费者”模型

这是一个交叉模型，无论是生产者还是消费者，都秉持同一个逻辑：

• 占在自己的锁上，条件不满足时一直等待。
• 一旦条件满足，开始工作。必要时可以获取公共资源的锁。
• 执行完任务，跑到对方的锁上唤醒对方的线程。

condition#await( ), condition#signalAll( )

除了wait()和notifyAll()来完成线程间的协作。conditon#await()和conditon#signalAll()也能实现同样的功能。

和wait()以及notifyAll()是附着于Object不同。conditon#await()和conditon#signalAll()是附着于Lock。

官方的例子：例子里通过两个条件来控制不同线程。

• Condition notFull：”防满溢标签“。当数组存满100个元素时，防满溢标签放出await()方法“阻塞，挂起，释放锁”。只有同一个标签放出signalAll()才能终止await()让线程继续。
• Condition notEmpty：”防空标签“。当数组中没有元素时，防空标签放出await()方法“阻塞，挂起，释放锁”。只有同一个标签放出signalAll()才能终止await()让线程继续。

class BoundedBuffer {
  final Lock lock = new ReentrantLock();
  final Condition notFull  = lock.newCondition(); 
  final Condition notEmpty = lock.newCondition(); 

  final Object[] items = new Object[100];
  int putptr, takeptr, count;

  public void put(Object x) throws InterruptedException {
    lock.lock();
    try {
      while (count == items.length)
        notFull.await();
      items[putptr] = x;
      if (++putptr == items.length) putptr = 0;
      ++count;
      notEmpty.signal();
    } finally {
      lock.unlock();
    }
  }

  public Object take() throws InterruptedException {
    lock.lock();
    try {
      while (count == 0)
        notEmpty.await();
      Object x = items[takeptr];
      if (++takeptr == items.length) takeptr = 0;
      --count;
      notFull.signal();
      return x;
    } finally {
      lock.unlock();
    }
  }
}

BlockingQueue

无论通过Object#wait(),notify()组合还是condition#await(),signal()组合，这种通过互斥锁握手来实现同步的策略还是有点复杂。

一个更简单的解决方案是BlockingQueue。它的特性主要有两点：

1. 对它的操作是“线程安全”的。所以它内部肯定是维护着一个互斥锁的。操作和操作之间具有原子性。可以放心地用。
2. 队列满了，插入操作会被阻塞挂起。空了，读取操作会被阻塞挂起。

然后通过各个例子来进一步加深理解和记忆吧，骚年：

（完）