Thread pool

What is a thread pool

Thread pool is a thread usage mode. The thread pool will maintain multiple threads waiting to allocate tasks that can be executed concurrently. When a task needs thread execution, it will allocate threads from the thread pool to the task without actively creating threads.

Benefits of thread pooling

If we usually need to use threads, we usually do this: create threads (T1), use the created threads to execute tasks (T2), and destroy the current thread (T3) after task execution. These three stages are necessary.

What if the thread pool is used?

The thread pool will create a certain number of threads in advance and apply for use when necessary. After a task is executed, it is not necessary to destroy the thread, which obviously saves the time of T1 and T3.

At the same time, our threads are uniformly managed by the thread pool, which also improves the manageability of threads.

Write your own thread pool

Now we can simply understand that thread pool is actually an array of multiple threads. When the program starts, a certain thread instance is instantiated in advance and allocated when there is a need for a task. Now let's write our own thread pool to understand the basic working process of a thread pool.

What does the thread pool need?

First of all, the thread pool must require a certain number of threads, so we first need an array of threads, of course, it can also be a collection.

Thread array is used to store thread instances. To use these threads, you need to submit tasks. When the number of tasks is too large, it is impossible to assign a thread to all tasks at the same time, so we also need a container for storing tasks.

The number of pre initialized thread instances here also needs to be determined according to the business.

At the same time, the number of thread instances cannot be defined arbitrarily, so we need to set a maximum number of threads.

//The maximum number of threads allowed in the thread pool
private static int MAXTHREDNUM = Integer.MAX_VALUE;
//The default number of threads when not specified by the user
private  int threadNum = 6;
//Thread queue to store thread tasks
private List<Runnable> queue;

private WorkerThread[] workerThreads;

Thread pool work

Threads in the thread pool generally need to be instantiated in advance. Here we simulate this process through the constructor.

 public MyThreadPool(int threadNum) {
        this.threadNum = threadNum;
        if(threadNum > MAXTHREDNUM)
            threadNum = MAXTHREDNUM;
        this.queue = new LinkedList<>();
        this.workerThreads = new WorkerThread[threadNum];
        init();
    }
//Initialize threads in the thread pool
 private void init(){
    for(int i=0;i<threadNum;i++){
        workerThreads[i] = new WorkerThread();
        workerThreads[i].start();
    }
}

After the thread pool is ready, we need to submit work tasks in the thread pool. The tasks are submitted to the queue uniformly. When there are tasks, threads are automatically distributed.

//Submit task
public void execute(Runnable task){
    synchronized (queue){
        queue.add(task);
        //Wake up the thread waiting in the queue after submitting the task
        queue.notifyAll();
    }
}

In order to obtain tasks, our working thread needs to monitor the task queue all the time. When there are tasks in the queue, a thread will execute them. Here we use the security interrupt mentioned above.

private class WorkerThread extends Thread {
private volatile boolean on = true;
@Override
public void run() {
    Runnable task = null;
    //Judge whether the task can be retrieved
    try {
        while(on&&!isInterrupted()){
            synchronized (queue){
                while (on && !isInterrupted() && queue.isEmpty()) {
                    //Here, if the blocking queue is used to obtain, there will be no error during execution
                    //An error is reported because all thread resources are destroyed when exiting, which does not affect the use
                    queue.wait(1000);
                }
                if (on && !isInterrupted() && !queue.isEmpty()) {
                    task = queue.remove(0);
                }

                if(task !=null){
                    //Execute after getting the task
                    task.run();
                }
            }
        }
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    task = null;//After the task is completed, manually empty it to accelerate recycling
}

public void cancel(){
    on = false;
    interrupt();
}
}

Of course, the threads in the thread pool need to be destroyed when exiting.

    //Destroy thread pool
    public void shutdown(){
        for(int i=0;i<threadNum;i++){
            workerThreads[i].cancel();
            workerThreads[i] = null;
        }
        queue.clear();
    }

Well, here we have completed a simple version of thread pool. Although there are few functions, the basic principle of thread pool operation is almost realized. In fact, it is very simple. Let's write a program to test it:

public class ThreadPoolTest {
    public static void main(String[] args) throws InterruptedException {
        // Create a thread pool of 3 threads
        MyThreadPool t = new MyThreadPool(3);
        CountDownLatch countDownLatch = new CountDownLatch(5);
        t.execute(new MyTask(countDownLatch, "testA"));
        t.execute(new MyTask(countDownLatch, "testB"));
        t.execute(new MyTask(countDownLatch, "testC"));
        t.execute(new MyTask(countDownLatch, "testD"));
        t.execute(new MyTask(countDownLatch, "testE"));
        countDownLatch.await();
        Thread.sleep(500);
        t.shutdown();// Destroy only after all threads have completed execution
        System.out.println("finished...");
    }

    // Task class
    static class MyTask implements Runnable {

        private CountDownLatch countDownLatch;
        private String name;
        private Random r = new Random();

        public MyTask(CountDownLatch countDownLatch, String name) {
            this.countDownLatch = countDownLatch;
            this.name = name;
        }

        public String getName() {
            return name;
        }

        @Override
        public void run() {// Perform tasks
            try {
                countDownLatch.countDown();
                Thread.sleep(r.nextInt(1000));
                System.out.println("task " + name + " complete");
            } catch (InterruptedException e) {
                System.out.println(Thread.currentThread().getId()+" sleep InterruptedException:"
                        +Thread.currentThread().isInterrupted());
            }
        }
    }
}

result: 
task testA complete
 task testB complete
 task testC complete
 task testD complete
 task testE complete
finished...
java.lang.InterruptedException
	at java.lang.Object.wait(Native Method)
	at com.learn.threadpool.MyThreadPool$WorkerThread.run(MyThreadPool.java:75)
...

After all the tasks are submitted, we can see that all the tasks are destroyed.