当前位置：网站首页>异步模式之工作线程

异步模式之工作线程

2022-07-28 23:01:00 【51CTO】

1. 定义

让有限的工作线程（Worker Thread）来轮流异步处理无限多的任务。也可以将其归类为分工模式，它的典型实现就是线程池，也体现了经典设计模式中的享元模式。

例如，海底捞的服务员（线程），轮流处理每位客人的点餐（任务），如果为每位客人都配一名专属的服务员，那么成本就太高了（对比另一种多线程设计模式：Thread-Per-Message）
注意，不同任务类型应该使用不同的线程池，这样能够避免饥饿，并能提升效率
例如，如果一个餐馆的工人既要招呼客人（任务类型A），又要到后厨做菜（任务类型B）显然效率不咋地，分成服务员（线程池A）与厨师（线程池B）更为合理，当然你能想到更细致的分工

2. 饥饿

        固定大小线程池会有饥饿现象
        ● 两个工人是同一个线程池中的两个线程
        ● 他们要做的事情是：为客人点餐和到后厨做菜，这是两个阶段的工作
                ○ 客人点餐：必须先点完餐，等菜做好，上菜，在此期间处理点餐的工人必须等待
                ○ 后厨做菜：没啥说的，做就是了
        ● 比如工人A 处理了点餐任务，接下来它要等着工人B 把菜做好，然后上菜，他俩也配合的蛮好
        ● 但现在同时来了两个客人，这个时候工人A 和工人B 都去处理点餐了，这时没人做饭了，饥饿

       
       public 
       
       class 
       
       TestDeadLock {
       
       static 
       
       final 
       
       List
       
       <
       
       String
       
       > 
       
       MENU 
       
       = 
       
       Arrays.
       
       asList(
       
       "地三鲜", 
       
       "宫保鸡丁", 
       
       "辣子鸡丁", 
       
       "烤鸡翅");
       
       static 
       
       Random 
       
       RANDOM 
       
       = 
       
       new 
       
       Random();
       
       static 
       
       String 
       
       cooking() {
       
       return 
       
       MENU.
       
       get(
       
       RANDOM.
       
       nextInt(
       
       MENU.
       
       size()));
       
    }
       
       public 
       
       static 
       
       void 
       
       main(
       
       String[] 
       
       args) {
       
       ExecutorService 
       
       executorService 
       
       = 
       
       Executors.
       
       newFixedThreadPool(
       
       2);
       
       executorService.
       
       execute(() 
       
       -> {    
       
       //excute执行一个任务，submit又提交一个任务，等于两个任务。一个线程处理点餐，另一个做菜
       
       //两个任务执行不会产生饥饿现象，但是多了就会产生饥饿。
       
       //两个线程都去处理点餐了，没有线程处理做菜了。f.get()无法获取值，陷入死等。但不是死锁。
       
       log.
       
       debug(
       
       "处理点餐...");
       
       Future
       
       <
       
       String
       
       > 
       
       f 
       
       = 
       
       executorService.
       
       submit(() 
       
       -> {
       
       log.
       
       debug(
       
       "做菜");
       
       return 
       
       cooking();
       
            });
       
       try {
       
       log.
       
       debug(
       
       "上菜: {}", 
       
       f.
       
       get());
       
            } 
       
       catch (
       
       InterruptedException 
       
       | 
       
       ExecutionException 
       
       e) {
       
       e.
       
       printStackTrace();
       
            }
       
        });
       
       /*executorService.execute(() -> {
       
       log.debug("处理点餐...");
       
       Future<String> f = executorService.submit(() -> {
       
       log.debug("做菜");
       
       return cooking();
       
       });
       
       try {
       
       log.debug("上菜: {}", f.get());
       
       } catch (InterruptedException | ExecutionException e) {
       
       e.printStackTrace();
       
       }
       
       });*/
       
    }
       
}
      
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.
37.

输出

17:21:27.883 c.TestDeadLock [pool-1-thread-1] - 处理点餐...
17:21:27.891 c.TestDeadLock [pool-1-thread-2] - 做菜
17:21:27.891 c.TestDeadLock [pool-1-thread-1] - 上菜: 烤鸡翅

当注释取消后，可能的输出

17:08:41.339 c.TestDeadLock [pool-1-thread-2] - 处理点餐...
17:08:41.339 c.TestDeadLock [pool-1-thread-1] - 处理点餐...

解决方法可以增加线程池的大小，不过不是根本解决方案，还是前面提到的，不同的任务类型，采用不同的线程池，例如：

       
       public 
       
       class 
       
       TestDeadLock {
       
       static 
       
       final 
       
       List
       
       <
       
       String
       
       > 
       
       MENU 
       
       = 
       
       Arrays.
       
       asList(
       
       "地三鲜", 
       
       "宫保鸡丁", 
       
       "辣子鸡丁", 
       
       "烤鸡翅");
       
       static 
       
       Random 
       
       RANDOM 
       
       = 
       
       new 
       
       Random();
       
       static 
       
       String 
       
       cooking() {
       
       return 
       
       MENU.
       
       get(
       
       RANDOM.
       
       nextInt(
       
       MENU.
       
       size()));
       
    }
       
       public 
       
       static 
       
       void 
       
       main(
       
       String[] 
       
       args) {
       
       ExecutorService 
       
       waiterPool 
       
       = 
       
       Executors.
       
       newFixedThreadPool(
       
       1);
       
       ExecutorService 
       
       cookPool 
       
       = 
       
       Executors.
       
       newFixedThreadPool(
       
       1);
       
       waiterPool.
       
       execute(() 
       
       -> {
       
       log.
       
       debug(
       
       "处理点餐...");
       
       Future
       
       <
       
       String
       
       > 
       
       f 
       
       = 
       
       cookPool.
       
       submit(() 
       
       -> {
       
       log.
       
       debug(
       
       "做菜");
       
       return 
       
       cooking();
       
            });
       
       try {
       
       log.
       
       debug(
       
       "上菜: {}", 
       
       f.
       
       get());
       
            } 
       
       catch (
       
       InterruptedException 
       
       | 
       
       ExecutionException 
       
       e) {
       
       e.
       
       printStackTrace();
       
            }
       
        });
       
       waiterPool.
       
       execute(() 
       
       -> {
       
       log.
       
       debug(
       
       "处理点餐...");
       
       Future
       
       <
       
       String
       
       > 
       
       f 
       
       = 
       
       cookPool.
       
       submit(() 
       
       -> {
       
       log.
       
       debug(
       
       "做菜");
       
       return 
       
       cooking();
       
            });
       
       try {
       
       log.
       
       debug(
       
       "上菜: {}", 
       
       f.
       
       get());
       
            } 
       
       catch (
       
       InterruptedException 
       
       | 
       
       ExecutionException 
       
       e) {
       
       e.
       
       printStackTrace();
       
            }
       
        });
       
    }
       
}
      
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.

输出

17:25:14.626 c.TestDeadLock [pool-1-thread-1] - 处理点餐...
17:25:14.630 c.TestDeadLock [pool-2-thread-1] - 做菜
17:25:14.631 c.TestDeadLock [pool-1-thread-1] - 上菜: 地三鲜
17:25:14.632 c.TestDeadLock [pool-1-thread-1] - 处理点餐...
17:25:14.632 c.TestDeadLock [pool-2-thread-1] - 做菜
17:25:14.632 c.TestDeadLock [pool-1-thread-1] - 上菜: 辣子鸡丁

因此，注意，不同任务类型应该使用不同的线程池，这样能够避免饥饿，并能提升效率。