第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
}     

5. 线程共享内存

多个线程可以对同一个变量进行读写操作，但由于读写是非原子性的以及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++;
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) 内存可见性

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

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缓存等原因，可能修改没有及时同步到内存，也可能另一个线程根本就没从内存读。

6. 线程的优点及成本

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

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

第二节 线程竞争

1. 线程竞争及解决方案(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实例方法。

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(实例方法中)或类对象(静态方法中)作为锁对象。

2. synchronized的特性

1) 可重入性

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

2) 内存可见性

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需进行加锁，当修饰代码块时，如果以不同顺序对多个对象进行加锁时，可能造成死锁。

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

5
    private static void startThreadA() {
6
        Thread aThread = new Thread() {
7

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

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

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

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

4) 无法响应中断

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

5) 只有一个条件队列

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

3. 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、没有迭代问题、直接支持一些复合操作、性能也高得多。

第三节 线程协作

多线程之间除了竞争，还经常需要相互协作，本节就来介绍下多线程协作的基本机制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。

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

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

5
1
synchronized (obj) {
2
    while (!条件成立)
3
        obj.wait();
4
    // 执行条件满足后的操作
5
}

接下来，我们通过一些场景来进一步理解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获取锁的

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

3. 使用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);
18
    
19
    // 4. 执行其他操作
20
    
21
    try {
22
        // 5. 获取异步调用的真实结果
23
        Integer result = future.get();
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
}

子线程启动后调用read方法，此时标准输入流无数据，会进行阻塞，虽然主线程设置了中断标记，但子线程并不能响应中断。如果输入一个字符，read方法执行通过，随后进入while循环的中断标志判断逻辑，此时发现被中断才会退出。

4. 如何处理线程中断

线程中断时，可能抛出InterruptedException异常，也可能只设置中断标识位。

• 对于前者，可以捕捉该异常，然后进行资源释放等一些操作，一般来说，还会重新设置中断标志位，方便后续程序处理。

• 对于后者，可以在循环代码中判断中断标志位，进行一系列的中断处理，如打印日志和退出循环等。

35
1
public class InterruptDemo {
2
    public static void main(String[] args) throws InterruptedException {
3
        // 1. 子线程
4
        Thread thread = new Thread(new Runnable() {
5
            @Override
6
            public void run() {
7
                // 1.1 执行业务代码
8
                while (true) {
9
                    // 业务代码
10
                    System.out.println("当前时间：" + System.currentTimeMillis());
11
                    try {
12
                        Thread.sleep(100); // 进入TIMED_WAITING状态
13
                    } catch (InterruptedException e) {
14
                        // 1.2 处理WAIT/TIMED_WAITING状态状态被中断
15
                        System.out.println("close...");
16
                        System.out.println(Thread.currentThread().isInterrupted()); // false，标志被重置
17
                        Thread.currentThread().interrupt(); // 重设中断标志位
18
                    }
19

20
                    // 1.3 处理Runnable状态被中断
21
                    if (Thread.currentThread().isInterrupted()) {
22
                        System.out.println("exit!");
23
                        break;
24
                    }
25
                }
26
            }
27
        });
28
        thread.start();
29

30
        // 2. 中断该子线程
31
        Thread.sleep(1000);
32
        thread.interrupt();
33
        thread.join();
34
    }
35
}

注意：

1. BLOCKED状态的线程虽然会被设置中断标识位，但依然保持阻塞状态，无法对其进行处理。

5. 如何取消/关闭线程

线程中断只是一种协作机制，而不一定会真正"中断"线程，如果不明白线程在做什么，不应该贸然的调用线程的interrupt方法。

那如何取消或关闭线程呢？一般而言，对于以线程提供服务的程序模块，它应该封装取消或关闭的操作，从而使调用者能正确关闭线程。

如上节的InterruptReadDemo，可以在子线程中提供如下cancel方法，调用后关闭流(注意，流在关闭时返回-1)，并进行中断退出。

8
1
// 取消
2
public void cancel() {
3
    try {
4
        System.in.close(); // 先关闭IO流
5
    } catch (IOException e) {
6
    }
7
    interrupt(); // 再中断当前线程
8
}

同样的，Java并发库的一些代码也提供了单独的取消/关闭方法：

6
1
// Future接口提供了如下方法以取消任务
2
boolean cancel(boolean mayInterruptIfRunning);
3

4
// ExecutorService提供了如下两个关闭方法
5
void shutdown();
6
List<Runnable> shutdownNow();

第02章_并发协作工具

第一节 原子变量

1. 基础用法

原子变量可以保证更新操作的原子性，且无需加锁以及上下文切换，效率更高。常用基本类型的原子变量及相关变体如下：