当前位置：网站首页>Asynchronous mode worker thread

Asynchronous mode worker thread

2022-07-29 00:49:00 【51CTO】

1. Definition

Let limited worker threads （Worker Thread） To take turns processing an infinite number of tasks asynchronously . It can also be classified as the mode of division of labor , Its typical implementation is thread pool , It also reflects the Heyuan mode in the classic design pattern .

for example , Undersea fishing waiter （ Threads ）, Take turns with each guest's order （ Mission ）, If each guest is equipped with a dedicated waiter , Then the cost is too high （ Contrast another multithreading design pattern ：Thread-Per-Message）
Be careful , Different task types should use different thread pools , It's a way to avoid starvation , And improve efficiency
for example , If a restaurant worker wants to greet guests （ Task type A）, I'm going to cook in the back kitchen again （ Task type B） Obviously not very efficient , It's divided into waiters （ Thread pool A） With the chef （ Thread pool B） More reasonable , Of course, you can think of a more detailed division of labor

2. hunger

         Fixed size thread pool will be hungry
        ● Two workers are two threads in the same thread pool
        ● What they're going to do is ： Ordering for guests and cooking in the back kitchen , It's two stages of work
                ○ The guests order ： You have to order first , Wait for the dishes to be ready , Serve , In the meantime, the workers handling the order have to wait
                ○ Back kitchen cooking ： Nothing to say. , Just do it
        ● Like workers A I've taken care of the order , Next it has to wait Worker B Do the dishes well , And then the dishes , They also cooperated very well
        ● But now there are two guests at the same time , At this time, the workers A And workers B We've all gone to order , At this time, no one cooked , hunger

       
       public 
       
       class 
       
       TestDeadLock {
       
       static 
       
       final 
       
       List
       
       <
       
       String
       
       > 
       
       MENU 
       
       = 
       
       Arrays.
       
       asList(
       
       " Sauteed Potato, Green Pepper and Eggplant ", 
       
       " Kung Pao Chicken ", 
       
       " Sauteed Chicken Dices with Chili Peppers ", 
       
       " Roast chicken wings ");
       
       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 Perform a task ,submit Submit another task , Equal to two tasks . A thread handles ordering , Another cook 
       
       // The execution of two tasks will not produce hunger , But more will lead to hunger .
       
       // Both threads are going to order , There are no threads to handle cooking .f.get() Unable to get value , Fall into death, etc . But it's not a deadlock .
       
       log.
       
       debug(
       
       " Handle order ...");
       
       Future
       
       <
       
       String
       
       > 
       
       f 
       
       = 
       
       executorService.
       
       submit(() 
       
       -> {
       
       log.
       
       debug(
       
       " cook a dish ");
       
       return 
       
       cooking();
       
            });
       
       try {
       
       log.
       
       debug(
       
       " Serve : {}", 
       
       f.
       
       get());
       
            } 
       
       catch (
       
       InterruptedException 
       
       | 
       
       ExecutionException 
       
       e) {
       
       e.
       
       printStackTrace();
       
            }
       
        });
       
       /*executorService.execute(() -> {
       
       log.debug(" Handle order ...");
       
       Future<String> f = executorService.submit(() -> {
       
       log.debug(" cook a dish ");
       
       return cooking();
       
       });
       
       try {
       
       log.debug(" Serve : {}", 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.

Output

17:21:27.883 c.TestDeadLock [pool-1-thread-1] - Handle order ...
17:21:27.891 c.TestDeadLock [pool-1-thread-2] - cook a dish
17:21:27.891 c.TestDeadLock [pool-1-thread-1] - Serve : Roast chicken wings

When the comment is cancelled , Possible outputs

17:08:41.339 c.TestDeadLock [pool-1-thread-2] - Handle order ...
17:08:41.339 c.TestDeadLock [pool-1-thread-1] - Handle order ...

The solution is to increase the size of the thread pool , But not the fundamental solution , As mentioned earlier , Different types of tasks , Use different thread pools , for example ：

       
       public 
       
       class 
       
       TestDeadLock {
       
       static 
       
       final 
       
       List
       
       <
       
       String
       
       > 
       
       MENU 
       
       = 
       
       Arrays.
       
       asList(
       
       " Sauteed Potato, Green Pepper and Eggplant ", 
       
       " Kung Pao Chicken ", 
       
       " Sauteed Chicken Dices with Chili Peppers ", 
       
       " Roast chicken wings ");
       
       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(
       
       " Handle order ...");
       
       Future
       
       <
       
       String
       
       > 
       
       f 
       
       = 
       
       cookPool.
       
       submit(() 
       
       -> {
       
       log.
       
       debug(
       
       " cook a dish ");
       
       return 
       
       cooking();
       
            });
       
       try {
       
       log.
       
       debug(
       
       " Serve : {}", 
       
       f.
       
       get());
       
            } 
       
       catch (
       
       InterruptedException 
       
       | 
       
       ExecutionException 
       
       e) {
       
       e.
       
       printStackTrace();
       
            }
       
        });
       
       waiterPool.
       
       execute(() 
       
       -> {
       
       log.
       
       debug(
       
       " Handle order ...");
       
       Future
       
       <
       
       String
       
       > 
       
       f 
       
       = 
       
       cookPool.
       
       submit(() 
       
       -> {
       
       log.
       
       debug(
       
       " cook a dish ");
       
       return 
       
       cooking();
       
            });
       
       try {
       
       log.
       
       debug(
       
       " Serve : {}", 
       
       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.

Output

17:25:14.626 c.TestDeadLock [pool-1-thread-1] - Handle order ...
17:25:14.630 c.TestDeadLock [pool-2-thread-1] - cook a dish
17:25:14.631 c.TestDeadLock [pool-1-thread-1] - Serve : Sauteed Potato, Green Pepper and Eggplant
17:25:14.632 c.TestDeadLock [pool-1-thread-1] - Handle order ...
17:25:14.632 c.TestDeadLock [pool-2-thread-1] - cook a dish
17:25:14.632 c.TestDeadLock [pool-1-thread-1] - Serve : Sauteed Chicken Dices with Chili Peppers

therefore , Be careful , Different task types should use different thread pools , It's a way to avoid starvation , And improve efficiency .

3. How many thread pools are appropriate to create

Too small will lead to the program can not make full use of system resources 、 It's easy to lead to hunger
Too much will lead to more thread context switching , Use more memory

3.1 CPU Intensive operations

Usually used cpu Check the number + 1 Be able to achieve optimal CPU utilization ,+1 Is to ensure that when the thread fails due to page loss （ operating system ） Or other reasons , The extra thread can be pushed up , Guarantee CPU Clock cycles are not wasted .

3.2 I/O Intensive operations

CPU Not always busy , for example , When you perform business calculations , At this time will use CPU resources , But when you execute I/O In operation 、 long-range RPC Invocation time , Including database operation , Now CPU I'm free , You can use multithreading to improve its utilization .

The empirical formula is as follows

Number of threads = Check the number * expect CPU utilization * Total time (CPU computing time + Waiting time ) / CPU computing time

for example 4 nucleus CPU The calculation time is 50% , Other waiting times are 50%, expect cpu By 100% utilize , To paraphrase formula ：4 * 100% * 100% / 50% = 8

for example 4 nucleus CPU The calculation time is 10% , Other waiting times are 90%, expect cpu By 100% utilize , To paraphrase formula ：4 * 100% * 100% / 10% = 40