2w字 + 40张图带你参透并发编程！

• 资源利用率，我们上面说到，单个进程存在资源浪费的情况，举个例子，当你在为某个文件夹赋予权限的时候，输入程序无法接受外部的输入字符，只有等到权限赋予完毕后才能接受外部输入。总的来讲，就是在等待程序时无法执行其他工作。如果在等待程序时可以运行另一个程序，那么将会大大提高资源的利用率。（资源并不会觉得累）因为它不会划水～
• 公平性，不同的用户和程序都能够使用计算机上的资源。一种高效的运行方式是为不同的程序划分时间片来使用资源，但是有一点需要注意，操作系统可以决定不同进程的优先级。虽然每个进程都有能够公平享有资源的权利，但是当有一个进程释放资源后的同时有一个优先级更高的进程抢夺资源，就会造成优先级低的进程无法获得资源，进而导致进程饥饿。
• 便利性，单个进程是是不用通信的，通信的本质就是信息交换，及时进行信息交换能够避免信息孤岛，做重复性的工作；任何并发能做的事情，单进程也能够实现，只不过这种方式效率很低，它是一种顺序性的。

并发和并行的关系

优势和劣势

并发为什么会出现

• CPU 使用缓存来中和和内存的访问速度差异
• 操作系统提供进程和线程调度，让 CPU 在执行指令的同时分时复用线程，让内存和磁盘不断交互，不同的 CPU 时间片 能够执行不同的任务，从而均衡这三者的差异
• 编译程序提供优化指令的执行顺序，让缓存能够合理的使用

线程带来的安全性问题

public class TSynchronized implements Runnable{

    static int i = 0;

    public void increase(){
        i++;
    }


    @Override
    public void run() {
        for(int i = 0;i < 1000;i++) {
            increase();
        }
    }

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

        TSynchronized tSynchronized = new TSynchronized();
        Thread aThread = new Thread(tSynchronized);
        Thread bThread = new Thread(tSynchronized);
        aThread.start();
        bThread.start();
        System.out.println("i = " + i);
    }
}

1. 可见性问题

2. 原子性问题

3. 有序性问题

4. 活跃性问题

• 互斥条件：指进程对所分配到的资源进行排它性使用，即在一段时间内某资源只由一个进程占用。如果此时还有其它进程请求资源，则请求者只能等待，直至占有资源的进程释放。
• 请求和保持条件：指进程已经保持至少一个资源，但又提出了新的资源请求，而该资源已被其它进程占有，此时请求进程阻塞，但又对自己已获得的其它资源保持占有。
• 不剥夺条件：指进程已获得的资源，在未使用完之前，不能被剥夺，只能在使用完时由自己释放。
• 循环等待：指在发生死锁时，必然存在一个进程对应的环形链。

while(true){...}

for(;;){}

5. 性能问题

引起线程切换的几种方式

• 当前正在执行的任务完成，系统的 CPU 正常调度下一个需要运行的线程
• 当前正在执行的任务遇到 I/O 等阻塞操作，线程调度器挂起此任务，继续调度下一个任务。
• 多个任务并发抢占锁资源，当前任务没有获得锁资源，被线程调度器挂起，继续调度下一个任务。
• 用户的代码挂起当前任务，比如线程执行 sleep 方法，让出CPU。
• 使用硬件中断的方式引起上下文切换

• 不要在多线程之间共享变量
• 将共享变量置为不可变的

什么是线程安全性

原子性

竞态条件

public class RaceCondition {
  
  private Signleton single = null;
  public Signleton newSingleton(){
    if(single == null){
      single = new Signleton();
    }
    return single;
  }
  
}

加锁机制

• 对方法进行加锁，确保多个线程中只有一个线程执行方法；
• 对某个对象实例（在我们上面的探讨中，变量可以使用对象来替换）进行加锁，确保多个线程中只有一个线程对对象实例进行访问；
• 对类对象进行加锁，确保多个线程只有一个线程能够访问类中的资源。

synchronized(lock){
  // 线程安全的代码
}

public class Retreent {
  
  public synchronized void doSomething(){
    doSomethingElse();
    System.out.println("doSomething......");
  }
  
  public synchronized void doSomethingElse(){
    System.out.println("doSomethingElse......");
}

并行 Worker

1. 并行 Worker 的优点

2. 并行 Worker 的缺点

流水线

1. 响应式 - 事件驱动系统

2. Actor 模型

3. Channels 模型

4. 流水线设计的优点

5. 流水线设计的缺点

• 通过继承 Thread 类来创建线程
• 通过实现 Runnable 接口来创建线程
• 通过 Callable 和 Future 来创建线程

继承 Thread 类来创建线程

public class TJavaThread extends Thread{

    static int count;

    @Override
    public synchronized void run() {
        for(int i = 0;i < 10000;i++){
            count++;
        }
    }

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

        TJavaThread tJavaThread = new TJavaThread();
        tJavaThread.start();
        tJavaThread.join();
        System.out.println("count = " + count);
    }
}

• 定义一个线程类使其继承 Thread 类，并重写其中的 run 方法，run 方法内部就是线程要完成的任务，因此 run 方法也被称为 执行体
• 创建了 Thread 的子类，上面代码中的子类是 TJavaThread
• 启动方法需要注意，并不是直接调用 run 方法来启动线程，而是使用 start 方法来启动线程。当然 run 方法可以调用，这样的话就会变成普通方法调用，而不是新创建一个线程来调用了。

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

  TJavaThread tJavaThread = new TJavaThread();
  tJavaThread.run();
  System.out.println("count = " + count);
}

使用 Runnable 接口来创建线程

public class TJavaThreadUseImplements implements Runnable{

    static int count;

    @Override
    public synchronized void run() {
        for(int i = 0;i < 10000;i++){
            count++;
        }
    }

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

        new Thread(new TJavaThreadUseImplements()).start();
        System.out.println("count = " + count);
    }

}

• 首先定义 Runnable 接口，并重写 Runnable 接口的 run 方法，run 方法的方法体同样是该线程的线程执行体。
• 创建线程实例，可以使用上面代码这种简单的方式创建，也可以通过 new 出线程的实例来创建，如下所示

TJavaThreadUseImplements tJavaThreadUseImplements = new TJavaThreadUseImplements();
new Thread(tJavaThreadUseImplements).start();

• 再调用线程对象的 start 方法来启动该线程。

使用 Callable 接口来创建线程

public class CallableTask implements Callable {

    static int count;
    public CallableTask(int count){
        this.count = count;
    }

    @Override
    public Object call() {
        return count;
    }

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

        FutureTask task = new FutureTask((Callable) () -> {
            for(int i = 0;i < 1000;i++){
                count++;
            }
            return count;
        });
        Thread thread = new Thread(task);
        thread.start();

        Integer total = task.get();
        System.out.println("total = " + total);
    }
}

• Callable 执行的任务有返回值，而 Runnable 执行的任务没有返回值
• Callable（重写）的方法是 call 方法，而 Runnable（重写）的方法是 run 方法。
• call 方法可以抛出异常，而 Runnable 方法不能抛出异常

使用线程池来创建线程

• 利用线程池能够复用线程、控制最大并发数。
• 实现任务线程队列缓存策略和拒绝机制。
• 实现某些与时间相关的功能，如定时执行、周期执行等。
• 隔离线程环境。比如，交易服务和搜索服务在同一台服务器上，分别开启两个线程池，交易线程的资源消耗明显要大；因此，通过配置独立的线程池，将较慢的交易服务与搜索服务隔开，避免服务线程互相影响。

new Thread(new(RunnableTask())).start()

// 替换为
  
Executor executor = new ExecutorSubClass() // 线程池实现类;
executor.execute(new RunnableTask1());
executor.execute(new RunnableTask2());

1. CachedThreadPool

public static void main(String[] args) {
  ExecutorService service = Executors.newCachedThreadPool();
  for(int i = 0;i < 5;i++){
    service.execute(new TestThread());
  }
  service.shutdown();
}

2. FixedThreadPool

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

3. SingleThreadExecutor

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

休眠

public class SuperclassThread extends TestThread{

    @Override
    public void run() {
        System.out.println(Thread.currentThread() + "starting ..." );

        try {
            for(int i = 0;i < 5;i++){
                if(i == 3){
                    System.out.println(Thread.currentThread() + "sleeping ...");
                    TimeUnit.MILLISECONDS.sleep(1000);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println(Thread.currentThread() + "wakeup and end ...");
    }

    public static void main(String[] args) {
        ExecutorService executors = Executors.newCachedThreadPool();
        for(int i = 0;i < 5;i++){
            executors.execute(new SuperclassThread());
        }
        executors.shutdown();
    }
}

优先级

public class SimplePriorities implements Runnable{

    private int priority;

    public SimplePriorities(int priority) {
        this.priority = priority;
    }

    @Override
    public void run() {
        Thread.currentThread().setPriority(priority);
        for(int i = 0;i < 100;i++){
            System.out.println(this);
            if(i % 10 == 0){
                Thread.yield();
            }
        }
    }

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

    public static void main(String[] args) {
        ExecutorService service = Executors.newCachedThreadPool();
        for(int i = 0;i < 5;i++){
            service.execute(new SimplePriorities(Thread.MAX_PRIORITY));
        }
        service.execute(new SimplePriorities(Thread.MIN_PRIORITY));
    }
}

作出让步

后台线程

public class SimpleDaemons implements Runnable{

    @Override
    public void run() {
        while (true){
            try {
                TimeUnit.MILLISECONDS.sleep(100);
                System.out.println(Thread.currentThread() + " " + this);
            } catch (InterruptedException e) {
                System.out.println("sleep() interrupted");
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        for(int i = 0;i < 10;i++){
            Thread daemon = new Thread(new SimpleDaemons());
            daemon.setDaemon(true);
            daemon.start();
        }
        System.out.println("All Daemons started");
        TimeUnit.MILLISECONDS.sleep(175);
    }
}

ThreadFactory

class SimpleThreadFactory implements ThreadFactory {
  public Thread newThread(Runnable r) {
    return new Thread(r);
  }
}

public interface ThreadFactory {

    // 构建一个新的线程。实现类可能初始化优先级，名称，后台线程状态和 线程组等
    Thread newThread(Runnable r);
}

public class DaemonThreadFactory implements ThreadFactory {

    @Override
    public Thread newThread(Runnable r) {
        Thread t = new Thread(r);
        t.setDaemon(true);
        return t;
    }
}

public class DaemonFromFactory implements Runnable{

    @Override
    public void run() {
        while (true){
            try {
                TimeUnit.MILLISECONDS.sleep(100);
                System.out.println(Thread.currentThread() + " " + this);
            } catch (InterruptedException e) {
                System.out.println("Interrupted");
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ExecutorService service = Executors.newCachedThreadPool(new DaemonThreadFactory());
        for(int i = 0;i < 10;i++){
            service.execute(new DaemonFromFactory());
        }
        System.out.println("All daemons started");
        TimeUnit.MILLISECONDS.sleep(500);
    }
}

public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
  return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                60L, TimeUnit.SECONDS,
                                new SynchronousQueue(),
                                threadFactory);
}

加入一个线程

public class TestJoinMethod extends Thread{

    @Override
    public void run() {
        for(int i = 0;i < 5;i++){
            try {
                TimeUnit.MILLISECONDS.sleep(1000);
            } catch (InterruptedException e) {
                System.out.println("Interrupted sleep");
            }
            System.out.println(Thread.currentThread() + " " + i);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        TestJoinMethod join1 = new TestJoinMethod();
        TestJoinMethod join2 = new TestJoinMethod();
        TestJoinMethod join3 = new TestJoinMethod();

        join1.start();
//        join1.join();

        join2.start();
        join3.start();
    }
}

线程异常捕获

public class ExceptionThread implements Runnable{

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

    public static void main(String[] args) {
        try {
            ExecutorService service = Executors.newCachedThreadPool();
            service.execute(new ExceptionThread());
        }catch (Exception e){
            System.out.println("eeeee");
        }
    }
}

public class ExceptionThread2 implements Runnable{

    @Override
    public void run() {
        Thread t = Thread.currentThread();
        System.out.println("run() by " + t);
        System.out.println("eh = " + t.getUncaughtExceptionHandler());
      
       // 手动抛出异常
        throw new RuntimeException();
    }
}

// 实现Thread.UncaughtExceptionHandler 接口，创建异常处理器
public class MyUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler{

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

public class HandlerThreadFactory implements ThreadFactory {

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

public class CaptureUncaughtException {

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