基础 | 并发编程 - [多线程进度控制]

CountDownLatch

字面直译
减数门闩

作用
让一些线程阻塞，直到另一些线程都执行完成

相当于一扇门，这个门上插了 n 根门闩，new CountDownLatch(n)
每个门闩有个人在往外拉（在干点什么任务的线程），没拉出来一根，门闩总数减 1
当所有门闩都拉出来，门闩总数是 0 了，就可以开门了

作用示意

使用示例

public static void main(String[] args) throws InterruptedException {
    CountDownLatch latch = new CountDownLatch(10);

    new Thread(()->{
        try {
            latch.await();
            System.out.println("A Sth on next step");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    },"A").start();
    new Thread(()->{
        try {
            latch.await();
            System.out.println("B Sth on next step");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    },"B").start();

    for(int i = 0; i < 10; i++) {
        new Thread(()->{
            try {
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName()+" over");
            latch.countDown();
        },String.valueOf(i)).start();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

CyclicBarrier

字面直译
循环屏障

作用
让一组线程达到一个屏障（同步点）时被阻塞，直到最后一个线程达到屏障
此后屏障才会放行，各个线程才能做后面的任务

作用示意

使用示例

public static void main(String[] args) {
    CyclicBarrier barrier = new CyclicBarrier(5,()->{System.out.println("All Is Ready");});

    for(int i = 0; i < 5; i++) {
        new Thread(()->{
            try {
                System.out.println(Thread.currentThread().getName()+" arrived");
                TimeUnit.MILLISECONDS.sleep(100);
                barrier.await();
            } catch (BrokenBarrierException | InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName()+" pass");
        },String.valueOf(i)).start();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

Semaphore

字面直译
信号灯

作用

• 使多个共享资源互斥使用
• 控制并发线程数

public static void main(String[] args) {
    Semaphore semaphore = new Semaphore(4);

    for(int i = 0; i < 7; i++) {
        new Thread(()->{
            try {
                semaphore.acquire();
                System.out.println(Thread.currentThread().getName()+" <-");
                TimeUnit.MILLISECONDS.sleep(200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }finally {
                System.out.println(Thread.currentThread().getName()+" ---->");
                semaphore.release();
            }
        }, String.valueOf(i)).start();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

ForkJoinPool / ForkJoinTask / RecursiveTask

public class ForkJoinDemo extends RecursiveTask<Integer>{
    private int start;
    private int end;

    public ForkJoinDemo(int start, int end) {
        this.start = start;
        this.end = end;
    }

    public static void main(String[] args) {

        System.out.println(new ForkJoinDemo(1,100).compute());
    }

    @Override
    protected Integer compute() {
        if (start == end)
            return start;

        if((end-start)>15){
            ForkJoinDemo demo =  new ForkJoinDemo(start,start+((end-start)>>1));//start+(end-start)>>1
            ForkJoinDemo demo2 =  new ForkJoinDemo(start+((end-start)>>1)+1,end);
            demo.fork();
            demo2.fork();
            return demo2.join()+ demo.join();
        }


        int sum = 0;
        for(int i = start;i<=end;i++){
            sum += i;
        }
        System.out.println(start+" ~ "+end+" = "+sum);
        return sum;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

CompletableFuture 使用

CompletableFuture 实现了 Future 和 CompletionStage 接口
CompletionStage 接口是对复杂工作中 每一个可完成的步骤 的抽象

常用方法
执行方法，可按是否有返回值、是否指定线程池归类

回调方法

获取结果方法

主动触发计算
complete()，尝试完成 get() 或 join() 方法
要求一个入参，调用后立即返回 get() 或 join() 方法是否被打断
未处理完时通过 get() 或 join() 方法或获取入参值

流程控制

• 流程控制方法可使前后步骤串行化
  通常用于处理下一个 CompletableStage，实现异步任务的顺序执行
• 下面方法都有 异步重载，如 thenRunAsync()
• 下面方法都有 异步指定线程池重载，如 thenRunAsync(Consumer , Executor)

关于异步和线程池的关系如下 (参考 示例 2)

• 定义 CompletableFuture 时
  • 若未指定线程池，通常使用默认的 ForkJoinPool
  • 若指定线程池，通常 使用指定的线程池
• 使用上述流程控制方法时
  • 若调用的流程控制方法，通常使用 与上一阶段一致的线程池
  • 若调用的流程控制方法的 异步重载，通常使用默认的 ForkJoinPool
  • 若调用的流程控制方法的 异步指定线程池重载，通常 使用指定的线程池
• 若某步骤执行极快，可能不使用线程池，直接使用调用时线程执行
• 比如下文 //1 的睡眠注释，会出现 main 线程
  这是因为根据线程调度优化，优化后不在切换上下文了，于是 supplyAsync 和后面的步骤可能直接使用调用线程

CompletableFuture 与其他多线程进度控制的区别

• CompletableFuture 的大部分功能可以通过其他进度控制实现
• CompletableFuture 更强调一个复杂流程中不同步骤的控制
  其他进度控制的子任务间更多的只是协同关系，而不是步骤的从属
• CompletableFuture 在接口设计上比其他进度控制强大
  它实现了 Future 、CompletionStage，因此同时具有这些接口和其调用者提供的功能

示例 1

public class CompletableFutureDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Void> f = CompletableFuture.runAsync(()->{
            System.out.println("==============");
        });
        f.get();

        System.out.println();
        System.out.println();
        CompletableFuture<Integer> f1 = CompletableFuture.supplyAsync(()->{
            System.out.println("**************");
            return 1;
        });
        f1.whenComplete((result,except)->{
            System.out.println(result);
    
            System.out.println(except);
        });
        System.out.println();
        System.out.println();
        CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync(()->{
            System.out.println("-------------");
            return 1/0;
        });
        f2.whenComplete((result,except)->{
            System.out.println(result);
            System.out.println(except);
        });
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

示例 2

public static void main(String[] args){
    CompletableFuture.supplyAsync(()->{
        // 1
        try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); }
        return 1;
    },Executors.newFixedThreadPool(3))

            .thenRun(()->{System.out.println("1 "+Thread.currentThread().getName());})
            .thenAcceptAsync(x->System.out.println("2 "+Thread.currentThread().getName()))
            .thenAcceptAsync(x->System.out.println("2 "+Thread.currentThread().getName()),Executors.newFixedThreadPool(3))
            .thenAcceptAsync(x->System.out.println("2 "+Thread.currentThread().getName()),Executors.newFixedThreadPool(3))
            .thenAccept(x->System.out.println("2 "+Thread.currentThread().getName()))
            .thenAccept(x->System.out.println("2 "+Thread.currentThread().getName()))

            .whenComplete((result, except)->{
        System.out.println(result);
        System.out.println(except);
    });
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

示例 3

public static void main(String[] args){
    CompletableFuture a = CompletableFuture.supplyAsync(()->{
        System.out.println("a start");
        try { TimeUnit.SECONDS.sleep(10); } catch (InterruptedException e) { e.printStackTrace(); }
        return "a";
    });
    CompletableFuture b = CompletableFuture.supplyAsync(()->{
        System.out.println("b start");
        try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); }
        return "b";
    });
    System.out.println(a.applyToEither(b,r-> r + " is applied : " + Thread.currentThread().getName()).join());
    System.out.println(a.thenCombine(b, (x,y)->x+" | "+y +": "+ Thread.currentThread().getName()).join());
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14