第03篇_并发编程

第01章_线程基础

第一节 线程的基本概念

1. 创建和启动线程

1) 直接自定义线程

可以通过继承java.lang.Thread类，重写其run方法来直接定义一个线程。

​x
1
// 1. 继承Thread类
2
public class HelloThread extends Thread {
3

4
    // 2. 重写run方法：线程入口，无参无返回值，不能抛出受检异常
5
    @Override
6
    public void run() {
7
        System.out.println("hello");
8
    }
9
}
10

11
public static void main(String[] args) {
12
    Thread thread = new HelloThread(); // 3. 创建线程对象
13
    thread.start(); // 4. 启动线程，分配程序计数器、栈等线程资源
14
}

2) 定义Runnable实现类

由于Java只支持单继承，因此一般都是定义java.lang.Runnable接口的实现类，然后创建Runnable的对象传给Thread对象去执行。

14
1
// 1. 实现Runnable接口
2
public class HelloRunnable implements Runnable {
3

4
    // 2. 重写run方法，该方法将优先于Thread的run方法进行调用
5
    @Override
6
    public void run() {
7
        System.out.println("hello");
8
    }
9
}
10

11
public static void main(String[] args) {
12
    Thread helloThread = new Thread(new HelloRunnable()); // 3. 创建Thread对象，并传递一个Runnable对象
13
    helloThread.start(); // 4. 启动线程
14
}

2. 线程基本属性

22
1
// 1. 获取当前线程
2
public static native Thread currentThread(); // Thread类静态方法
3

4
// 2. 线程ID和名称
5
public long getId(); // 线程ID，一个递增的整数，每创建一个线程就加一
6
public final String getName(); // 线程名称，默认值是"Thread-"后跟一个编号
7
public final synchronized void setName(String name) // 设置线程名称(也可以在创建Thread对象时指定)
8
    
9
// 3. 线程优先级
10
public final int getPriority() // 获取线程优先级，从1~10，默认为5
11
public final void setPriority(int newPriority) // 设置线程优先级，映射到不同操作系统优先级上(仅供提示)
12
    
13
// 4. 线程状态
14
public State getState() // 获取线程状态
15
    
16
// 5. 线程存活状态
17
public final native boolean isAlive() // 线程只有在NEW和TERMINATED状态才为false，其它都为true
18

19
// 6. 是否为守护线程(daemo线程，即其它线程的辅助线程，如垃圾回收线程等，当整个程序只剩下daemo线程时，程序就会退出。）
20
public final boolean isDaemon() // 是否为守护线程，默认为false
21
public final void setDaemon(boolean on) // 设置为守护线程
22

3. 线程六大状态

线程状态为Thread.State枚举，有如下六种状态：

注意：

1. RUNNABLE不代表CPU一定在执行该线程的代码，也可能在等待操作系统分配时间片，只是它没有在等待其他条件。

4. 线程基本操作

21
1
// 1. 睡眠指定毫秒。时间有一定的偏差，在睡眠期间，会让出CPU，可以被其他线程中断(中断后抛出InterruptedException)
2
public static native void sleep(long millis) throws InterruptedException; // Thread的静态方法
3

4
// 2. 让出CPU(仅供提示)。
5
public static native void yield(); // Thread的静态方法
6

7
// 3. 等待该线程结束。转为WAIT/TIMED_WAITING状态，等待期间可以被其他线程中断
8
public final void join() throws InterruptedException // 内部调用了join(0)
9
public final synchronized void join(long millis) throws InterruptedException 
10

11
// 4. 一些过时的方法，不应该去使用。
12
public final void stop()
13
public final void suspend()
14
public final void resume()
15
    
16
// 示例1：主线程等待thread线程，先不要退出
17
public static void main(String[] args) throws InterruptedException {
18
    Thread thread = new HelloThread();
19
    thread.start();
20
    thread.join();
21
}     

注意：

1. wait() 是对象级别的方法，定义在 Object 类，用于线程间的协作，需要在同步代码块中调用，会释放对象锁。

2. join() 是线程级别的方法，定义在 Thread 类，用于线程的同步，不会释放锁，用于等待目标线程执行完成。

5. 线程的优点及成本

优点：多线程可以充分利用多核CPU的计算能力以及相关的硬件资源，因为一个线程在占用CPU时，另一个线程可以同时进行IO等操作。同理，由于多个执行流，对保持程序的响应性也十分有帮助，如GUI程序中，一个线程处理慢任务时，另一个线程还能够向用户反馈进度信息。

成本：创建和启动线程需要消耗操作系统的资源，并且，线程调度和切换也是有成本的。因此，如果执行的任务都是CPU密集型的，即主要消耗的都是CPU，那创建超过CPU数量的线程就是没有必要的，并不会加快程序的执行。

6. 面试补充

1) 进程 vs 线程？

• 进程：进程是操作系统分配资源的最小单位，在启动Java应用时，就是启动了一个JVM进程，它包括主线程(main)和其它线程。

• 线程：线程是CPU调度和执行的最小单位，线程之间共享进程的堆和方法区(元空间)，但是有各自的方法栈和程序计数器等，更适合任务协作和高并发性能的场景。

2) 并行 vs 并发

• 并行：两个及两个以上的作业在同一 时间段 内执行。

• 并发：两个及两个以上的作业在同一 时刻执行，多核CPU才可能并发。

3) 如何创建线程？

• Java中只有一种方式可以创建和执行线程，那就是：new Thread().start()。

• 但是可以通过继承Thread类、实现Runnable/Callable接口、使用线程池、使用CompletableFuture类等间接调用。

4) 多线程的优点和缺点

• 优点：多线程可提高IO密集型应用的执行效率，提升GUI程序的响应性等。

• 缺点：提升了编码复杂性，可能带来线程安全问题或死锁等。

5) 并发编程的三大特性

• 有序性：由于指令重排序问题，代码的执行顺序未必就是编写代码时候的顺序（指令重排序可以保证串行语义一致，但是没有义务保证多线程间的语义也一致 ），在 Java 中，volatile 关键字可以禁止指令进行重排序优化。

• 可见性：当一个线程对共享变量进行了修改，那么另外的线程都是立即可以看到修改后的最新值，在 Java 中，可以借助volatile 、synchronized以及各种 Lock 实现可见性。

• 原子性：一次操作或者多次操作，要么所有的操作全部都得到执行并且不会受到任何因素的干扰而中断，要么都不执行，在 Java 中，可以借助synchronized、各种 Lock 以及各种原子类实现原子性。。

6) Thread#sleep() vs Object#wait()

• Thread#sleep：暂停当前线程，让出CPU，但不释放锁。

• Object#wait()：进入等待状态，让出CPU，且释放锁，需要被其它线程唤醒或超时后自动唤醒，常用于线程间协作。

第二节 线程竞争

1. 线程竞争情形

多个线程可以对同一个变量进行读写操作，但由于非原子性读写或CPU缓存等原因，可能形成竞态条件或出现内存可见性问题。

1) 竞态条件

竞态条件(race condition)指当多个线程读写同一个对象时，最终执行结果与执行时序有关，可能正确，也可能不正确。

29
1
public class CounterThread extends Thread {
2
    private static int counter = 0; // 1. 共享变量counter，初始值为0
3

4
    @Override
5
    public void run() {
6
        // 2. 每个线程对其自增1000次
7
        for (int i = 0; i < 1000; i++) {
8
            counter++; // 非原子性，等效 counter = counter + 1;
9
        }
10
    }
11

12
    public static void main(String[] args) throws InterruptedException {
13
        // 3. 创建和启动1000个线程
14
        int num = 1000;
15
        Thread[] threads = new Thread[num];
16
        for (int i = 0; i < num; i++) {
17
            threads[i] = new CounterThread();
18
            threads[i].start();
19
        }
20

21
        // 4. 等待所有线程结束
22
        for (int i = 0; i < num; i++) {
23
            threads[i].join();
24
        }
25

26
        // 5. 发现执行结果基本都<1000*1000次
27
        System.out.println(counter); // 998964/998469/999082
28
    }
29
}

为什么会这样呢？因为counter++不是原子操作，它先取counter的当前值，然后在当前值的基础上加1，再赋值回counter，而在它取到当前值到赋新值之间，当前值也可能被其它线程取走。

2) 内存可见性问题

内存可见性问题指一个线程对共享变量的修改，由于CPU缓存原因，另一个线程不一定马上就能看到，甚至永远也看不到。

26
1
public class VisibilityDemo {
2
    private static boolean shutdown = false; // 1. 共享变量shutdown，初始为false
3

4
    static class HelloThread extends Thread {
5
        @Override
6
        public void run() {
7
            // 2. 当shutdown为true时，子线程退出并输出"exit hello"
8
            while (!shutdown) {
9
                // do nothing
10
            }
11
            System.out.println("exit hello");
12
        }
13
    }
14

15
    public static void main(String[] args) throws InterruptedException {
16
        // 3. 启动子线程并将shutdown改为true，发现子线程并没有退出
17
        HelloThread helloThread = new HelloThread();
18
        helloThread.start();
19
        Thread.sleep(1000);
20
        shutdown = true;
21
        
22
        // 4. 无论是否等待子线程，子线程都不会发现shutdown已经改为true了
23
        // helloThread.join();
24
        System.out.println("exit main");
25
    }
26
}

为什么会这样呢？由于CPU缓存等原因，可能修改没有及时同步到内存，也可能另一个线程根本就没从内存读。

2. 线程竞争解决方案(synchronized)

前面说到，当多个线程读写同一个变量时，可能形成竞态条件或出现内存可见性问题，这就是线程之间的竞争，在Java中，最简单的解决方式是使用synchronized关键字。

1) 修饰实例方法

上节我们介绍竞态条件时，了解到counter++是非原子操作，可能造成结果比预期偏小，现在我们可以对counter其进行包装，通过synchronized关键字修饰的实例方法来对其进行自增和读取操作，这样就始终和预期一致了。

13
1
public class Counter {
2
    private int count; // 共享变量
3

4
    // synchronized
5
    public synchronized void incr() {
6
        count++;
7
    }
8

9
    // synchronized
10
    public synchronized int getCount() {
11
        return count;
12
    }
13
}

这里，synchronized到底做了什么呢？在进入synchronized实例方法前，会尝试对this对象进行加锁操作，当前线程不能获得锁的时候，它会进入等待队列(此时线程状态变为BLOCKED)，这样来确保只有一个线程执行。

注意：

1. synchronized是对this对象加锁，而非对方法加锁。

   • 当this对象不同时，是可以执行同一个synchronized实例方法的；

   • 当this对象相同时，即使是其它synchronized实例方法也是不可以执行的。

2. synchronized机制不能防止非synchronized方法被同时执行，如Counter类添加一个非synchronized的decr方法，进行count--操作，是无法同步的，因此，一般在保护变量时，需要在所有访问该变量的方法上加上synchronized。

2) 修饰静态方法

synchronized同样可以用于静态方法，比如：

11
1
public class StaticCounter {
2
    private static int count = 0; // 共享变量
3

4
    public static synchronized void incr() {
5
        count++;
6
    }
7

8
    public static synchronized int getCount() {
9
        return count;
10
    }
11
}

当修饰静态方法时，不能对this对象加锁了，它将对类对象(StaticCounter.class)进行加锁，加锁逻辑基本一致。

注意：

1. synchronized静态方法和synchronized实例方法对不同对象进行加锁，也就是说，不同的两个线程，可以一个执行synchronized静态方法，另一个执行synchronized实例方法。

2. synchronized不可以修饰构造方法（因为不可能多个线程同时调用同一个对象的构造方法），如需在构造时访问共享资源，只能在构造方法内部修饰代码块。

3) 修饰代码块

除了用于修饰方法外，synchronized还可以用于修饰代码块，比如对于前面的Counter类，等价的代码可以为：

16
1
public class Counter {
2
    private int count; // 共享变量
3
    private Object lock = new Object(); // 锁对象(lock也可以是this或类对象)
4
    
5
    public void incr(){
6
        synchronized(lock){ 
7
            count ++;    
8
        }
9
    }
10
    
11
    public int getCount() {
12
        synchronized(lock){
13
            return count;
14
        }
15
    }
16
} 

此时，加锁的对象就是小括号里面的lock变量，它可以是任意对象。但是一般来说，在使用synchronized修饰代码块时，习惯使用this(实例方法中)或类对象(静态方法中)作为锁对象。

注意：

1. 尽量不要使用包装类或字符串作为锁对象，因为它们有缓存池或常量池。

2. 使用 字符串.intern() 作为锁对象时，由于字符串常量池的存在，锁对象粒度会过大。

3. synchronized的特性

1) 可重入性

synchronized方法或代码块是可重入的，即某个线程获得锁之后，再次调用synchronized方法可直接调用，它是通过记录锁的持有线程和持有数量来实现的。

2) 内存可见性(volatile)

synchronized还可以保证共享变量的内存可见性，在获得锁后，会始终从内存读取最新数据，在释放锁后，所有写入都会写回内存。

但如果只需要保证内存可见性，而无需保证原子性，可以使用更轻量级的方式，那就是给共享变量加修饰符volatile。

11
1
public class Switcher {
2
    private volatile boolean on; // 共享变量，加volatile保证内存可见性
3

4
    public boolean isOn() {
5
        return on;
6
    }
7

8
    public void setOn(boolean on) {
9
        this.on = on;
10
    }
11
}

加了volatile之后，Java会在操作对应变量时插入特殊的指令，保证读写到内存最新值，而非缓存的值。

3) 可能死锁

synchronized需进行加锁，当修饰代码块时，如果以不同顺序对多个对象进行加锁时，可能造成死锁。

50
1
public class DeadLockDemo {
2
    private static Object lockA = new Object(); // 全局锁A
3
    private static Object lockB = new Object(); // 全局锁B
4

5
    // A线程
6
    private static void startThreadA() {
7
        Thread aThread = new Thread() {
8

9
            @Override
10
            public void run() {
11
                // 1. 线程A：先持有锁A，等待会后去申请锁B
12
                synchronized (lockA) {
13
                    try {
14
                        Thread.sleep(1000);
15
                    } catch (InterruptedException e) {
16
                    }
17
                    synchronized (lockB) {
18
                    }
19
                }
20
            }
21
        };
22
        aThread.start();
23
    }
24

25
    // B线程
26
    private static void startThreadB() {
27
        Thread bThread = new Thread() {
28
            @Override
29
            public void run() {
30
                // 2. 线程B：先持有锁B，等待会后去申请锁A
31
                synchronized (lockB) {
32
                    try {
33
                        Thread.sleep(1000);
34
                    } catch (InterruptedException e) {
35
                    }
36
                    synchronized (lockA) {
37
                    }
38
                }
39
            }
40
        };
41
        bThread.start();
42
    }
43

44
    public static void main(String[] args) {
45
        // 3. 启动两个线程，可能造成死锁(线程A等线程B释放B锁，线程B等待线程A释放A锁)
46
        startThreadA();
47
        startThreadB();
48
    }
49
}
50

死锁可通过jstack -pid命令查看，我们在写程序时，应该尽量避免在持有一个锁的同时去申请另一个锁，如果确实需要多个锁，所有代码都应该按照相同的顺序去申请锁。在Java中，也有一些显式锁支持tryLock以及加锁超时的设定，也可以一定程度解决死锁问题。

4) 无法响应中断

获取synchronized锁时，如果未获取到，将进入BLOCK状态，该状态无法响应中断请求，而显式锁Lock则支持以响应中断的方式获取锁。

5) 只有一个条件队列

synchronized的协作机制wait/notify，只支持一个条件队列，如果有多个等待条件，只能共用，并且在通知时必须通知所有等待的线程。

4. synchronized容器

1) 基本使用

Collections类有一些方法可以修饰普通容器，返回线程安全的同步容器，它们是给所有容器方法都加上synchronized来实现安全的。

3
1
public static <T> Collection<T> synchronizedCollection(Collection<T> c)
2
public static <T> List<T> synchronizedList(List<T> list)
3
public static <K,V> Map<K,V> synchronizedMap(Map<K,V> m)

这里线程安全针对的是容器对象，指的是当多个线程并发访问同一个容器对象时，不需要额外的同步操作，也不会出现错误的结果。

2) 复合操作与伪同步

加了synchronized，所有方法调用变成了原子操作，客户端在调用时，是不是就绝对安全了呢？不是的，至少有如下情况需要注意：

19
1
// 自定义Map
2
public class EnhancedMap <K, V> {
3
    Map<K, V> map;
4
    
5
    public EnhancedMap(Map<K,V> map){
6
        this.map = Collections.synchronizedMap(map); // 1. 修饰为同步的Map
7
    }
8
    
9
    // 2. 期望实现putIfAbsent的原子操作
10
    public V putIfAbsent(K key, V value){
11
         V old = map.get(key);
12
         if(old!=null){
13
             return old;
14
         }
15
         map.put(key, value);
16
         return null;
17
     }
18
}
19

上述案例中的原子操作显然是不能实现的，虽然map.get(key)和map.put(key, value)是原子的，但是它们组合后就不是原子的了。

那给putIfAbsent方法加上synchronized呢？也同样也不能实现原子操作，因为putIfAbsent方法的synchronized是对当前的EnhancedMap对象加锁，而map.get(key)和map.put(key, value)方法是对map加锁，这是一种伪同步。

但我们可以在putIfAbsent方法中使用synchronized代码块，并且使用map进行加锁。

11
1
public V putIfAbsent(K key, V value){
2
    // synchronized，使用map进行加锁
3
    synchronized(map){
4
        V old = map.get(key);
5
         if(old!=null){
6
             return old;
7
         }
8
         map.put(key, value);
9
         return null;    
10
    }
11
}

3) 并发修改异常

此外，还需注意，虽然同步容器的单个操作是安全的，但是在增删元素时进行迭代操作依然可能会抛出 ConcurrentModificationException。

38
1
private static void startModifyThread(final List<String> list) {
2
    Thread modifyThread = new Thread(new Runnable() {
3
        @Override
4
        public void run() {
5
            // 1. 线程1：增删list的元素
6
            for (int i = 0; i < 100; i++) {
7
                list.add("item " + i);
8
                try {
9
                    Thread.sleep((int) (Math.random() * 10));
10
                } catch (InterruptedException e) {
11
                }
12
            }
13
        }
14
    });
15
    modifyThread.start();
16
}
17

18
private static void startIteratorThread(final List<String> list) {
19
    Thread iteratorThread = new Thread(new Runnable() {
20
        @Override
21
        public void run() {
22
            // 2. 线程2：重复进行遍历操作
23
            while (true) {
24
                for (String str : list) {
25
                }
26
            }
27
        }
28
    });
29
    iteratorThread.start();
30
}
31

32
public static void main(String[] args) {
33
    // 3. 创建同步容器list进行测试，发现会抛出ConcurrentModificationException
34
    final List<String> list = Collections.synchronizedList(new ArrayList<String>());
35
    startIteratorThread(list);
36
    startModifyThread(list);
37
}
38

同步容器并没有解决这个问题，如果要避免这个异常，需要在遍历的时候给整个容器对象加锁，如startIteratorThread可以改为：

16
1
private static void startIteratorThread(final List<String> list) {
2
    Thread iteratorThread = new Thread(new Runnable() {
3
        @Override
4
        public void run() {
5
            // 2. 线程2：重复进行遍历操作
6
            while (true) {
7
                // 但是在遍历前对list进行加锁
8
                synchronized(list){
9
                    for (String str : list) {
10
                    }    
11
                }
12
            }
13
        }
14
    });
15
    iteratorThread.start();
16
}

4) 性能问题

除了以上这些注意事项，同步容器的性能也是比较低的，当并发访问量比较大的时候性能很差。在高并发场景，可以使用CopyOnWriteArrayList、ConcurrentHashMap、ConcurrentLinkedQueue、ConcurrentSkipListSet等专为并发设计的容器类，它们都是线程安全的，但都没有使用synchronized、没有迭代问题、直接支持一些复合操作、性能也高得多。

5. 面试补充

1) volatile

• volatile关键字可修饰变量，表示该变量是共享且不稳定的，禁用 CPU 缓存，每次都从主内存读取，防止出现内存可见性问题。

• 另外，还会通过插入特定的 内存屏障 的方式来防止 JVM 的指令重排序。

  19
  1
  // 使用双检锁方式创建单例对象
  2
  public class Singleton {
  3
      private volatile static Singleton uniqueInstance; // 必须要加 volatile 防止指令重排
  4
      private Singleton() {
  5
      }
  6
      public  static Singleton getUniqueInstance() {
  7
          if (uniqueInstance == null) {
  8
              synchronized (Singleton.class) {
  9
                  if (uniqueInstance == null) {
  10
                      // 1) 为 uniqueInstance 分配内存空间
  11
                      // 2) 初始化 uniqueInstance
  12
                      // 3) 将 uniqueInstance 指向分配的内存地址
  13
                      uniqueInstance = new Singleton(); // 如果不使用volatile，可能会先执行1和3，被其它线程获取到尚未初始化的实例
  14
                  }
  15
              }
  16
          }
  17
          return uniqueInstance;
  18
      }
  19
  }
  

• volatile关键字不能保证对变量的操作是原子性的。

2) synchronized

• synchronized是最常用的同步手段之一，可修饰实例方法、静态方法、代码块等。

• synchronized底层原理：

  • 修饰代码块：通过对象监视器monitor来实现。

  • 修饰方法：对方法打 ACC_SYNCHRONIZED 标识，标识需要同步调用。

• synchronized锁升级：

  • 同步对象初始是无锁状态，当一个线程访问同步块并获取锁时，会升级为偏向锁（该锁在高版本JDK已被禁用）。

  • 当其它线程也试图访问该对象时，偏向锁会升级为轻量级锁，获取到锁的线程执行，其它线程进行自旋。

  • 当锁竞争加剧时，自旋会大量消耗CPU资源，此时会升级为重量级锁，获取到锁的线程执行，其它线程阻塞等待。

第三节 线程协作

多线程之间除了竞争，还经常需要相互协作，本节就来介绍下多线程协作的基本机制wait/notify。

1. 线程协作的场景

多线程之间需要协作的场景有很多，比如说：

• 生产者/消费者协作模式：生产者线程和消费者线程通过共享队列进行协作，生产者将数据或任务放到队列上，而消费者从队列上取数据或任务，如果队列长度有限，在队列满的时候，生产者需要等待，而在队列为空的时候，消费者需要等待。

• 同时开始：类似运动员比赛，在听到比赛开始枪响后同时开始，在一些程序，尤其是模拟仿真程序中，要求多个线程能同时开始。

• 等待结束：主从协作模式也是一种常见的协作模式，主线程将任务分解为若干个子任务，为每个子任务创建一个线程，主线程在继续执行其他任务之前需要等待每个子任务执行完毕。

• 集合点：比如并行迭代计算中，每个线程负责一部分计算，然后在集合点等待其他线程完成，所有线程到齐后，交换数据和计算结果，再进行下一次迭代。

• 异步结果：在主从协作模式中，主线程手工创建子线程的写法往往比较麻烦，一种常见的模式是将子线程的管理封装为异步调用，异步调用马上返回，但返回的不是最终的结果，而是一个一般称为Promise或Future的对象，通过它可以在随后获得最终的结果。

我们会探讨如何实现这些协作场景，在此之前，我们先来了解协作的基本方法wait/notify。

2. wait/notify机制

在Java中，任意对象都可以参与线程协作，它们都有wait和notify方法，可以在synchronized代码块中调用：

7
1
// 1. wait
2
public final void wait() throws InterruptedException // 无限期等待，等同wait(0)
3
public final native void wait(long timeout) throws InterruptedException // 超时等待，timeout表示等待时间，单位为毫秒
4

5
// 2. notify
6
public final native void notify(); // 通知，从条件队列中选一个线程，将其从队列中移除并唤醒
7
public final native void notifyAll(); // 通知所有，移除条件队列中所有的线程并全部唤醒

前面提到，当线程等待synchronized锁时，会将线程加入到锁对象的等待队列中。而线程在synchronized代码块内主动调用wait方法时，会将线程加入到另一个条件队列，并进行阻塞。

调用wait方法后，将释放持有的锁，线程状态变为WAITING/TIMED_WAITING，当被唤醒后，需重新竞争锁。

• 如果获取到锁，线程状态变为RUNNABLE，并从wait调用中返回。

• 如果未获取到锁，将会进入等待队列，线程状态变为BLOCKED，直到获得锁后才能从wait调用返回。

被阻塞的线程需要其它线程调用该对象的notify方法来将它唤醒。调用notify后，不会立即释放持有的锁，而是等待synchronzied代码块执行完后才释放。这意味着wait方法必须等到调用notify的那段同步代码执行结束后才可能获得锁响应通知。

注意：

1. wait/notify方法只能在synchronized代码块内被调用，否则将抛出IllegalMonitorStateException（无对象监视器：monitor）。

2. join方法也可进行等待，如前面介绍的主线程等待子线程结束案例，当子线程运行结束时，由java系统调用notifyAll来通知。

在使用wait/notify时，一定要明确wait等待的是什么？在满足什么条件后调用notify？并且需注意，从wait返回后，不一定表示等待的条件就满足了，仍需进行条件检查，因此，wait方法的一般调用模式为：

7
1
synchronized (obj) {
2
    // 循环检查条件是否满足
3
    while (!条件成立)
4
        obj.wait(); // 不满足则释放锁，进入等待
5
    
6
    // 执行条件满足后的操作
7
}

接下来，我们通过一些场景来进一步理解wait/notify的应用。

3. 生产者/消费者模式

在生产者/消费者模式中，协作的共享变量是队列，生产者往队列上放数据，如果满了就wait，而消费者从队列上取数据，如果队列为空也wait。我们将队列作为单独的类进行设计，代码如下：

28
1
static class MyBlockingQueue<E> {
2
    private Queue<E> queue = null; // 1. 共享变量：队列
3
    private int limit; // 队列上限
4

5
    public MyBlockingQueue(int limit) {
6
        this.limit = limit;
7
        queue = new ArrayDeque<>(limit);
8
    }
9

10
    // 2. 生产者调用，放数据，如果满了则等待
11
    public synchronized void put(E e) throws InterruptedException {
12
        while (queue.size() == limit) {
13
            wait(); // 等待(注意，锁对象为this)
14
        }
15
        queue.add(e);
16
        notifyAll(); // 4. 生产数据后可通知消费者“有数据了”！(注意：这里只能通知所有线程)
17
    }
18

19
    // 3. 消费者调用，拿数据，空了也等待
20
    public synchronized E take() throws InterruptedException {
21
        while (queue.isEmpty()) {
22
            wait();
23
        }
24
        E e = queue.poll();
25
        notifyAll(); // 5. 拿走数据后可通知生产者“有空位了”(注意：这里只能通知所有线程)
26
        return e;
27
    }
28
}

在上面代码中，生产者和消费者都调用了wait方法，但它们等待的条件是不一样的。生产者在队列为满时等待，而消费者在队列为空的时候等待，它们等待条件不同但又使用相同的条件队列，所以要调用notifyAll而不能调用notify，因为notify可能唤醒的恰好是同类线程。

类似的，它们也都调用了notifyAll方法，但是需满足的条件也不一致，生产者在“有数据了”的条件下通知消费者，消费者在“有空位了”的条件下通知生产者。

注意：synchronized的局限性

1. synchronized获取锁的wait/notify协作机制，只能有一个条件等待队列，这是它的局限性，使的分析变得复杂，后面将会介绍支持多个条件的显式锁。

2. 使用synchronized关键字获取锁的过程中不响应中断请求，这也是synchronized的局限性。

4. 同时开始

在同时开始模式中，协作的共享变量是一个开始信号。我们用一个类FireFlag来表示这个协作对象：

16
1
static class FireFlag {
2
    private volatile boolean fired = false; // 1. 共享变量：是否开始
3

4
    // 2. 子线程在启动后调用，等待开始
5
    public synchronized void waitForFire() throws InterruptedException {
6
        while (!fired) {
7
            wait(); // 等待(注意，锁对象为this)
8
        }
9
    }
10

11
    // 3. 主线程调用，发起开始指令
12
    public synchronized void fire() {
13
        this.fired = true; 
14
        notifyAll(); // 通知所有wait的线程
15
    }
16
}

子线程应该调用waitForFire()等待枪响，而主线程应该调用fire()发射比赛开始信号。代码如下所示：

34
1
// 子线程
2
static class Racer extends Thread {
3
    FireFlag fireFlag; // 1. 共享变量
4

5
    public Racer(FireFlag fireFlag) {
6
        this.fireFlag = fireFlag;
7
    }
8

9
    @Override
10
    public void run() {
11
        try {
12
            // 2. 等待开始
13
            this.fireFlag.waitForFire();
14
            System.out.println("start run " + Thread.currentThread().getName());
15
        } catch (InterruptedException e) {
16
        }
17
    }
18
}
19

20
// 主线程
21
public static void main(String[] args) throws InterruptedException {
22
    // 1. 启动20个子线程
23
    int num = 10;
24
    FireFlag fireFlag = new FireFlag();
25
    Thread[] racers = new Thread[num];
26
    for (int i = 0; i < num; i++) {
27
        racers[i] = new Racer(fireFlag);
28
        racers[i].start();
29
    }
30
    Thread.sleep(1000);
31
    
32
    // 2. 开始
33
    fireFlag.fire(); 
34
}

5. 等待结束

在等待结束模式中，主线程与各个子线程协作的共享变量是一个数，这个数表示未结束线程个数。

22
1
public class MyLatch {
2
    private int count; // 1. 共享变量
3

4
    public MyLatch(int count) {
5
        this.count = count; // 初始化为子线程个数
6
    }
7

8
    // 2. 主线程调用，等待所有子线程结束
9
    public synchronized void await() throws InterruptedException {
10
        while (count > 0) {
11
            wait();
12
        }
13
    }
14

15
    // 3. 子线程在退出前调用，线程计数 - 1
16
    public synchronized void countDown() {
17
        count--;
18
        if (count <= 0) {
19
            notifyAll();
20
        }
21
    }
22
}

应用代码示例如下：

38
1
// 子线程
2
static class Worker extends Thread {
3
    MyLatch latch;
4

5
    public Worker(MyLatch latch) {
6
        this.latch = latch;
7
    }
8

9
    @Override
10
    public void run() {
11
        try {
12
            // 1. 执行业务
13
            Thread.sleep((int) (Math.random() * 1000));
14

15
            // 2. 结束前调用countDown，线程计数-1
16
            this.latch.countDown(); 
17
        } catch (InterruptedException e) {
18
        }
19
    }
20
}
21

22
// 主线程
23
public static void main(String[] args) throws InterruptedException {
24
    // 1. 初始化计数
25
    int workerNum = 100;
26
    MyLatch latch = new MyLatch(workerNum); 
27
    
28
    // 2. 启动100个线程
29
    Worker[] workers = new Worker[workerNum];
30
    for (int i = 0; i < workerNum; i++) {
31
        workers[i] = new Worker(latch);
32
        workers[i].start();
33
    }
34
    
35
    // 3. 等待子线程都结束
36
    latch.await(); 
37
    System.out.println("collect worker results");
38
}

注意：

1. 可将线程计数初始值设置为1，由子线程调用await()，主线程调用countDown()，还可实现上面的“同时开始”模式。

7. 集合点

在集合点模式，协作的共享变量依然是一个数，这个数表示未到集合点的线程个数。

28
1
// CyclicBarrier
2
public class AssemblePoint {
3
    private int n; // 1. 共享变量
4

5
    public AssemblePoint(int n) {
6
        this.n = n;
7
    }
8

9
    // 2. 集合，线程达到集合点时调用，等待其它未到的线程。
10
    public synchronized void await() throws InterruptedException {
11
        // 2.1 n >0 表示还有线程未到
12
        if (n > 0) {
13
            // 2.2 线程计数-1
14
            n--;
15
            
16
            // 2.3 n==0 表示当前线程为最后一个到达线程，此时唤醒其它线程
17
            if (n == 0) {
18
                notifyAll();
19
            } else {
20
                // 2.3 n>0 表示还有线程未到，进行等待
21
                while (n != 0) {
22
                    wait();
23
                }
24
            }
25
            
26
        }
27
    }
28
}

多个游客线程，各自先独立运行，然后使用该协作对象到达集合点进行同步的示例代码如下：

37
1
// 需集合的线程
2
public class AssemblePointDemo {
3
    static class Tourist extends Thread {
4
        AssemblePoint ap; // 1. 共享变量
5

6
        public Tourist(AssemblePoint ap) {
7
            this.ap = ap;
8
        }
9

10
        @Override
11
        public void run() {
12
            try {
13
                // 2. 模拟先各自独立运行
14
                Thread.sleep((int) (Math.random() * 1000));
15

16
                // 3. 集合
17
                ap.await();
18
                System.out.println("arrived");
19
                
20
                // 4. 集合后执行其他操作
21
            } catch (InterruptedException e) {
22
            }
23
        }
24
    }
25

26
    // 主线程
27
    public static void main(String[] args) {
28
        // 1. 启动多个需集合的线程
29
        int num = 10;
30
        Tourist[] threads = new Tourist[num];
31
        AssemblePoint ap = new AssemblePoint(num);
32
        for (int i = 0; i < num; i++) {
33
            threads[i] = new Tourist(ap);
34
            threads[i].start();
35
        }
36
    }
37
}

6. 异步结果

异步结果模式依赖异步调用框架，主要由调用者、执行器、异步任务、异步结果四个部分组成。

其中异步任务和异步结果代码表示如下：

9
1
// 异步任务
2
public interface Callable<V> {
3
    V call() throws Exception;
4
}
5

6
// 异步结果
7
public interface MyFuture <V> {
8
    V get() throws Exception ; // 返回真正的结果。如果尚未计算完成，则等待。如果计算过程发生了异常，则抛出保存的异常。
9
}

执行器用于执行子任务并返回异步结果，使用执行器后调用者就无需创建并管理子线程了，其代码如下：

78
1
// 执行器
2
public <V> MyFuture<V> execute(final Callable<V> task) {
3
    // 1. 创建锁对象
4
    final Object lock = new Object(); 
5
    
6
    // 2. 创建和启动执行线程
7
    final ExecuteThread<V> thread = new ExecuteThread<>(task, lock);
8
    thread.start();
9

10
    // 3. 创建异步结果对象，用于获取异步结果
11
    MyFuture<V> future = new MyFuture<V>() {
12
        @Override
13
        public V get() throws Exception {
14
            synchronized (lock) {
15
                
16
                // 3.1 执行线程未结束，则等待
17
                while (!thread.isDone()) {
18
                    try {
19
                        lock.wait();
20
                    } catch (InterruptedException e) {
21
                    }
22
                }
23
                
24
                // 3.2 执行过程中出现异常，在调用get时直接抛出异常
25
                if (thread.getException() != null) {
26
                    throw thread.getException(); 
27
                }
28
                
29
                // 3.3 执行线程完结束后，直接返回保存的真实结果
30
                return thread.getResult(); 
31
            }
32
        }
33
    };
34
    
35
    // 4. 返回一个异步结果对象
36
    return future;
37
}
38

39
// 执行器委托给执行线程去执行
40
static class ExecuteThread<V> extends Thread {
41
    private V result = null; // 结果
42
    private Exception exception = null; // 异常
43
    private boolean done = false; // 是否完成
44
    private Callable<V> task; // 任务
45
    private Object lock; // 锁对象
46
    
47
    public ExecuteThread(Callable<V> task, Object lock) {
48
        this.task = task;
49
        this.lock = lock;
50
    }
51

52
    @Override
53
    public void run() {
54
        // 1. 执行任务，捕获异常，修改完成标记，通知其它线程
55
        try {
56
            result = task.call();
57
        } catch (Exception e) {
58
            exception = e;
59
        } finally {
60
            synchronized (lock) {
61
                done = true;
62
                lock.notifyAll();
63
            }
64
        }
65
    }
66

67
    public V getResult() {
68
        return result;
69
    }
70

71
    public boolean isDone() {
72
        return done;
73
    }
74

75
    public Exception getException() {
76
        return exception;
77
    }
78
}

调用者只需创建执行器，然后执行异步任务，即可得到异步结果对象。

28
1
public static void main(String[] args) {
2
    // 1. 创建执行器
3
    MyExecutor executor = new MyExecutor();
4
    
5
    // 2. 创建异步任务
6
    Callable<Integer> subTask = new Callable<Integer>() {
7
        @Override
8
        public Integer call() throws Exception {
9
            // 执行异步任务
10
            int millis = (int) (Math.random() * 1000);
11
            Thread.sleep(millis);
12
            return millis;
13
        }
14
    };
15
    
16
    // 3. 执行异步任务，返回异步结果
17
    MyFuture<Integer> future = executor.execute(subTask); // MyFuture执行的匿名内部类中对象中封装了执行线程对象
18
    
19
    // 4. 执行其他操作
20
    
21
    try {
22
        // 5. 获取异步调用的真实结果
23
        Integer result = future.get(); // 等待执行线程的finally中调用 lock.notifyAll() 
24
        System.out.println(result); // 6.1 处理真实结果
25
    } catch (Exception e) {
26
        e.printStackTrace(); // 6.2 处理异步任务异常
27
    }
28
}

第四节 线程中断

在Java中，停止一个线程的主要机制是中断，中断并不是强迫终止一个线程，而是一种协作机制，是给线程传递一个取消信号，但是由线程来决定如何以及何时退出，本节我们主要就是来理解Java的中断机制。

1. 线程中断的场景

通过线程的start方法启动一个线程后，线程开始执行run方法，run方法运行结束后线程退出，但有些场景需要对其进行中断：

• 线程的运行模式是死循环，无法自己停止。如在前面生产者/消费者模式案例中，生产者的代码就是一个死循环，不断的生产数据。

• 希望中途取消该线程执行的任务。如从远程服务器下载一个文件，在下载过程中，用户可能会希望取消该任务。

• 希望在限定的时间内得到结果。如从第三方服务器查询一个结果，超过设定的时间后仍未查询结束，则关闭查询。

• 通知取消其他渠道。如让多个好友帮忙抢火车票，只要有一个渠道抢到了，则通知其他人取消抢票。

2. 线程中断相关方法

每个线程都有一个中断标志位，表示该线程是否被中断，可通过如下方式获取和设置：

6
1
// 1. Thread类的实例方法
2
public boolean isInterrupted() // 获取线程是否被中断，即获取中断标志位
3
public void interrupt() // 中断该线程，即设置中断标志位为true
4

5
// 2. Thread类的静态方法
6
public static boolean interrupted() // 检查当前线程是否被中断(设置了中断标识位)，注意：此操作也会同时清空中断标志位，以方便后续处理。

注意：

1. 线程中还定义了stop()等一些方法，但它们已经过时了，不应该再继续使用。

3. 线程对中断的反应

线程被中断时，如何进行响应与线程的状态及进行的IO操作有关。

此外，重点介绍下InputStream的read调用，在流中没有数据时，read会阻塞 ，但线程依然是RUNNABLE状态。如果此时被中断，只会设置线程中断标志，并不能真正“中断”它。

27
1
public class InterruptReadDemo {
2
    // 子线程
3
    private static class A extends Thread {
4
        @Override
5
        public void run() {
6
            while(!Thread.currentThread().isInterrupted()){
7
                try {
8
                    System.out.println(System.in.read()); // read，在流中无数据时阻塞
9
                } catch (IOException e) {
10
                    e.printStackTrace();
11
                }    
12
            }
13
            System.out.println("exit");
14
        }
15
    }
16

17
    // 调用者
18
    public static void main(String[] args) throws InterruptedException {
19
        // 启动线程A
20
        A t = new A();
21
        t.start();
22
        Thread.sleep(100);
23

24
        // 想要中断线程A，实际上并不会被中断
25
        t.interrupt();
26
    }
27
}