当前位置：网站首页>[multithreading] use the thread pool to implement a simple thread pool

[multithreading] use the thread pool to implement a simple thread pool

2022-06-30 19:36:00 【51CTO】

Catalog

Why introduce thread pool
Why is the thread pool released faster than the system request ？
Java The thread pool in the standard library uses

ThreadPoolExecutor
Executors

Thread pools execute processes
Implement thread pool manually
Code

Why introduce thread pool

We know that every time we create start-up The cost of destroying a thread is large Therefore, the biggest advantage of introducing thread pool is to reduce the number of threads per startup Destroy thread loss

So how did he achieve the reduction ?

There is a blocking queue in the thread pool He will record and store the tasks to be performed In addition, there will be one or several threads inside the queue to fetch the first element of the random task

Just a quick example ：

Namely There is an express station An express The boss of the express station will hire a student to deliver the express Then immediately fire people Another express, he will hire another student And fired At this point he will find It's too much trouble to hire and dismiss like this So he hired a student or a student Then take a notebook to record the future express delivery Let several students take turns to see the express to be delivered in this notebook And then send them to the express This is a thread pool

Why is the thread pool released faster than the system request ？

To understand the problem , We have to understand a set of important concepts .

stay operating system in , We are divided into two states ：
1、 User mode
2、 Kernel mode

【 Multithreading 】 Use thread pool 、 Implement a simple thread pool _ Thread pool

In order to deepen “ Kernel mode ” and “ User mode ” The understanding of the .

Take an example of image ：

【 Multithreading 】 Use thread pool 、 Implement a simple thread pool _ Thread pool _02

below , Let's take a look Java What is embodied in the standard library .

Old rules , Let's learn Java Make sure you are in the library , The use of thread pools .
Then implement a thread pool by yourself .

Java The thread pool in the standard library uses

ThreadPoolExecutor

Create a thread pool in which parameters are understood

public ThreadPoolExecutor(int corePoolSize,
        int maximumPoolSize,
        long keepAliveTime,
        TimeUnit unit,
        BlockingQueue<Runnable> workQueue,
        ThreadFactory threadFactory,
        RejectedExecutionHandler handler)

     1.
2.
3.
4.
5.
6.
7.

corePoolSize: Number of core threads
maximumPoolSize: Maximum number of threads
keepAliveTime: Amount of free time
unit: A unit of free time
workQueue: Blocking / Mission / Work queue
threadFactory: Thread creation factory
handler: Refusal strategy

ThreadPoolExecutor

【 Multithreading 】 Use thread pool 、 Implement a simple thread pool _ Implement thread pool _03

Executors

A simplified version of thread pool is also provided in the standard library ----》Executors
Executors： The essence is for ThreadPoolExecutor It was packaged , Some default parameters are provided .

Let's take a look now Executors How does it work ？ By the way, we simulate a thread pool according to this .

【 Multithreading 】 Use thread pool 、 Implement a simple thread pool _java_04

Let's use the first newFixedThreadPool Method . To accomplish a task

【 Multithreading 】 Use thread pool 、 Implement a simple thread pool _ Thread pool _05

Thread pools execute processes

【 Multithreading 】 Use thread pool 、 Implement a simple thread pool _java_06

According to the scene above , We can imagine ourselves as the boss of the express station , Just imagine , If someone comes and sends something , Then I'll hire a temporary worker , When it's done , Immediately dismiss him , Yes, yes , But it doesn't seem very cost-effective , After all, the cost of hiring and firing temporary workers frequently begins to be relatively large .

brainwave , I decided to follow the actual needs of my shop , Employ 3 Regular employees , If all three of them are busy , Then I will put the package to be sent in my shop first , When the three of them have finished handling their parcels , Just continue to deal with the parcels piled up in the store , The cost is not very big , The days are light ~

But I didn't think ,618 The activity caught me by surprise , Soon my shop was full of , Three regular employees can't handle it , What if someone still comes to send the package at this time , Don't break the sender's heart ~ To show our strength , Small as the shop is , You can still stand a few more people , Don't panic , I wish I could hire more temporary workers , Site restrictions , My shop can only stand at most 6 personal , That is, I can only hire 3 A temporary worker , It's comfortable now .

But good days don't last long , Or am I too young , Underestimated the strength of consumers , Soon , Parcels piled up in the shop , Watching the six of them busy , I also feel bad , here , And a steady stream of people came to send packages , this time , I really can't help it , The heart that should be hurt still needs to be hurt , The spirit is willing but the flesh is weak , I had to refuse and tell them to find another place ~

After being a boss, you have to take it easy , Compare the flow chart above to understand the process pool ~ The core thread corresponds to 3 A regular employee , The blocking queue corresponds to the parcel storage area of the express store , The thread pool corresponds to the front desk of the store ( At most 6 personal ).

Implement thread pool manually

What's in the thread pool ？

1、 First be able to Describe task ------> Use it directly Runnable that will do
2、 need Organizational tasks ----- > Use it directly BlockingQueue Blocking queues
3、 Describe worker threads
4、 Organize worker threads
5、 Need to achieve , Add tasks to the thread pool .

The thread pool I want to implement , Not just managing tasks and Threads , They also need to cooperate with each other .

There is a class that describes what a specific thread does ( With the help of Runnable Realization )

static class Command implements Runnable {
        private int num;
        public Command(int num) {
            this.num = num;
        }

        @Override
        public void run() {
            System.out.println(" Executing task : " + num);
        }
    }


     1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.

A blocking team is needed to record several missions BlockingQueue

//  It is essentially a producer consumer model .
    //  call  execute  Your code is the producer .  Production tasks  (Runnable  object )
    // worker  Threads are consumers .  Consumed the tasks in the queue .
    //  The trading floor is  BlockingQueue
    static class MyThreadPool2 {
        //  This blocking queue is used to organize several tasks 
        private BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
        //  This  List  Used to organize several worker threads 
        private List<Worker> workers = new ArrayList<>();
        //  How many threads should there be in a thread pool ,  It needs to be determined according to the actual situation .
        //  Currently written  "10"  It's just a slap on the forehead 
        private static final int maxWorkerCount = 10;

        //  Realization  execute  Method   and  shutdown  Method 
        public void execute(Runnable command) throws InterruptedException {
            //  It also uses the method of delayed loading to create threads .
            //  When the number of threads in the thread pool is small ,  Create a new thread as a worker thread .
            //  If the number of threads is already large ( The set threshold is reached ),  There is no need to create a new thread .
            if (workers.size() < maxWorkerCount) {
                Worker worker = new Worker(queue, workers.size());
                worker.start();
                workers.add(worker);
            }
            queue.put(command);
        }

        //  When  shutdown  After that ,  It means that all threads must have ended .
        public void shutdown() throws InterruptedException {
            //  Terminate all threads .
            for (Worker worker : workers) {
                worker.interrupt();
            }
            //  You also need to wait for each thread to finish executing .
            for (Worker worker : workers) {
                worker.join();
            }
        }
    }

     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.
38.

Need a class to work on implementing threads

//  Use this class to describe what the current worker thread looks like .
    static class Worker extends Thread {
        private int id = 0;
        //  Every  Worker  Threads need to get tasks from the task queue .
        //  You need to be able to get the instance of the task queue 
        private BlockingQueue<Runnable> queue = null;

        public Worker(BlockingQueue<Runnable> queue, int id) {
            this.queue = queue;
            this.id = id;
        }

        @Override
        public void run() {
            //  Pay attention to  try  hold  while  The package is in .
            //  The purpose is that as long as the thread receives an exception ,  It will end immediately  run  Method ( That is, end the thread )
            try  {
                while (!Thread.currentThread().isInterrupted()) {
                    Runnable command = queue.take();
                    System.out.println("thread " + id + " running...");
                    command.run();
                }
            } catch (InterruptedException e) {
                //  The thread was terminated .
                System.out.println(" Thread terminated ");
            }
        }
    }

     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.

You need a linked list to store all threads For our convenience shutdown All threads

   //  When  shutdown  After that ,  It means that all threads must have ended .
        public void shutdown() throws InterruptedException {
            //  Terminate all threads .
            for (Worker worker : workers) {
                worker.interrupt();
            }
            //  You also need to wait for each thread to finish executing .
            for (Worker worker : workers) {
                worker.join();
            }
        }

     1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.

Code

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class TeacterThreadPool {
    //  Use this class to describe what the current worker thread looks like .
    static class Worker extends Thread {
        private int id = 0;
        //  Every  Worker  Threads need to get tasks from the task queue .
        //  You need to be able to get the instance of the task queue 
        private BlockingQueue<Runnable> queue = null;

        public Worker(BlockingQueue<Runnable> queue, int id) {
            this.queue = queue;
            this.id = id;
        }

        @Override
        public void run() {
            //  Pay attention to  try  hold  while  The package is in .
            //  The purpose is that as long as the thread receives an exception ,  It will end immediately  run  Method ( That is, end the thread )
            try  {
                while (!Thread.currentThread().isInterrupted()) {
                    Runnable command = queue.take();
                    System.out.println("thread " + id + " running...");
                    command.run();
                }
            } catch (InterruptedException e) {
                //  The thread was terminated .
                System.out.println(" Thread terminated ");
            }
        }
    }

    //  It is essentially a producer consumer model .
    //  call  execute  Your code is the producer .  Production tasks  (Runnable  object )
    // worker  Threads are consumers .  Consumed the tasks in the queue .
    //  The trading floor is  BlockingQueue
    static class MyThreadPool2 {
        //  This blocking queue is used to organize several tasks 
        private BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
        //  This  List  Used to organize several worker threads 
        private List<Worker> workers = new ArrayList<>();
        //  How many threads should there be in a thread pool ,  It needs to be determined according to the actual situation .
        //  Currently written  "10"  It's just a slap on the forehead 
        private static final int maxWorkerCount = 10;

        //  Realization  execute  Method   and  shutdown  Method 
        public void execute(Runnable command) throws InterruptedException {
            //  It also uses the method of delayed loading to create threads .
            //  When the number of threads in the thread pool is small ,  Create a new thread as a worker thread .
            //  If the number of threads is already large ( The set threshold is reached ),  There is no need to create a new thread .
            if (workers.size() < maxWorkerCount) {
                Worker worker = new Worker(queue, workers.size());
                worker.start();
                workers.add(worker);
            }
            queue.put(command);
        }

        //  When  shutdown  After that ,  It means that all threads must have ended .
        public void shutdown() throws InterruptedException {
            //  Terminate all threads .
            for (Worker worker : workers) {
                worker.interrupt();
            }
            //  You also need to wait for each thread to finish executing .
            for (Worker worker : workers) {
                worker.join();
            }
        }
    }

    static class Command implements Runnable {
        private int num;
        public Command(int num) {
            this.num = num;
        }

        @Override
        public void run() {
            System.out.println(" Executing task : " + num);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        MyThreadPool2 pool = new MyThreadPool2();
        for (int i = 0; i < 1000; i++) {
            pool.execute(new Command(i));
        }
        Thread.sleep(2000);
        pool.shutdown();
        System.out.println(" The thread pool has been destroyed ");
    }
}



     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.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.
66.
67.
68.
69.
70.
71.
72.
73.
74.
75.
76.
77.
78.
79.
80.
81.
82.
83.
84.
85.
86.
87.
88.
89.
90.
91.
92.
93.
94.
95.
96.
97.
98.
99.
100.