多线程之异步模式工作线程

1 定义

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

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(() -> {
         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:08:41.339 c.TestDeadLock [pool-1-thread-2] - 处理点餐... 
17:08:41.339 c.TestDeadLock [pool-1-thread-1] - 处理点餐... 
1
2

可以通过增加线程池大小解决问题, 但是不是根本解决问题, 因为来了比线程池数量更多的任务,还是会出现饥饿问题. 根本解决方案是: 不同的任务类型,使用不同的线程池,即点菜的任务, 从点菜线程池中获取; 做菜的任务,从做菜的线程池中获取.

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
37

运行结果:

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] - 上菜: 辣子鸡丁
1
2
3
4
5
6

3 线程池创建参考

线程池创建大小,直接影响程序执行效率,过大或过小都会带来问题.

• 过小会导致程序不能充分地利用系统资源、容易导致饥饿
• 过大会导致更多的线程上下文切换，占用更多内存

1 CPU 密集型运算

通常采用 cpu 核数 + 1 能够实现最优的 CPU 利用率，+1 是保证当线程由于页缺失故障（操作系统）或其它原因 导致暂停时，额外的这个线程就能顶上去，保证 CPU 时钟周期不被浪费

2 I/O 密集型运算

CPU 不总是处于繁忙状态，如当你执行业务计算时，这时候会使用 CPU 资源，当执行 I/O 操作时、远程 RPC 调用时，包括进行数据库操作时，这时 CPU 就空闲，可以利用多线程提高它的利用率。

数量计算公式:

线程数 = 核数 * 期望 CPU 利用率 * 总时间(CPU计算时间+等待时间) / CPU 计算时间

如 8 核 CPU 计算时间是 50% ，其它等待时间是 50%，期望 cpu 被 100% 利用，套用公式

8 * 100% * 100% / 50% = 16

根据经验公式, 上面线程池数量设置为16个.