手写线程池实战

Posted 2021-07-17 李子捌

有经典，有干货，微信搜索【李子捌】关注这个傻瓜式坚持的程序员。

往期回顾-线程池技术详解

线程池实战

简介：
上文讲述了怎么写一个简单的数据库连接池，试想一下，当服务端接收到来自客户端的成白上千个连接请求时，如果每次我们都为其创建一个线程去执行任务，执行完毕后销毁线程，这肯定不是一个好的解决办法。我们知道线程的创建和销毁是要消耗系统资源的，那么我们就可以从这个方面入手来控制资源的浪费。
​

线程池描述：
线程池技术是一种池化思想，也就是提前预置若干数量的线程，并且线程不能由用户直接控制，在这个前提下重复使用固定数量或者可以特定条件下进行伸缩的线程数目来完成任务的执行。
​

线程池的优点：

1. 消除频繁创建和消亡线程带来的系统开销
2. 面对过量任务提交时能够平缓的劣化

代码示例：
1、接口定义
描述，接口定义了提交任务到线程池执行的方法execute(Job job)，也提供增大/减少工作者线程和关闭线程的方法。工作者指的是，执行客户端提交任务的真实工作线程，工作者线程会从一个工作队列中获取等待执行的Job进行处理。

package com.lizba.p3.threadpool;

/**
 * <p>
 *      线程池接口
 * </p>
 *
 * @Author: Liziba
 * @Date: 2021/6/17 22:28
 */
public interface ThreadPool<Job extends Runnable> {

    /**
     * 执行一个Job，这个Job需要实现Runnable
     * @param job
     */
    void execute(Job job);

    /**
     * 关闭线程池
     */
    void shutdown();

    /**
     * 增加工作者线程
     * @param num
     */
    void addWorkers(int num);

    /**
     * 减少工作者线程
     * @param num
     */
    void removeWorkers(int num);

    /**
     * 得到正在等待执行的任务数量
     * @return
     */
    int getJobSize();
}

2、默认实现

package com.lizba.p3.threadpool;

import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

/**
 * <p>
 *		线程池默认实现
 * </p>
 *
 * @Author: Liziba
 * @Date: 2021/6/17 22:34
 */
public class DefaultThreadPool<Job extends Runnable> implements ThreadPool<Job> {

    /** 线程池最大工作者线程数量 */
    private static final int MAX_WORKER_SIZE = 20;
    /** 线程池默认工作者线程数量 */
    private static final int DEFAULT_WORKER_SIZE = 5;
    /** 线程池最小工作者线程数量 */
    private static final int MIN_WORKER_SIZE = 5;
    /** 工作队列，也称任务队列，用来存放客户端提交的任务 */
    private final LinkedList<Job> jobs = new LinkedList<>();
    /** 工作者列表，需要具有同步性质，支持并发操作 */
    private final List<Worker> workers = Collections.synchronizedList(new ArrayList<Worker>());
    /** 工作线程的数量 */
    private int workerNum = DEFAULT_WORKER_SIZE;
    /** 线程编号生成器 */
    private AtomicLong threadNum = new AtomicLong();

    public DefaultThreadPool() {
        initWorker(DEFAULT_WORKER_SIZE);
    }

    /**
     * 初始化线程工作者，并启动
     *
     * @param size  初始化工作着大小
     */
    private void initWorker(int size) {
        for (int i = 0; i < size; i++) {
            Worker worker = new Worker();
            workers.add(worker);
            Thread thread = new Thread(worker, "ThreadPool-Worker-" + threadNum.incrementAndGet());
            thread.start();
        }
    }

    @Override
    public void execute(Job job) {
        if (job != null) {
            // 添加一个任务，然后通知等待在jobs上的worker
            synchronized (jobs) {
                jobs.add(job);
                jobs.notifyAll();
            }
        }
    }

    @Override
    public void shutdown() {
        workers.forEach(worker -> worker.shutdown());
    }

    @Override
    public void addWorkers(int num) {
        // 此处要锁住jobs，因为worker会从jobs获取任务，需要jobs通知等待中的worker
        synchronized (jobs) {
            // 不允许工作者线程数操作最大值
            if (num + this.workerNum > MAX_WORKER_SIZE) {
                num = MAX_WORKER_SIZE - this.workerNum;
            }
            initWorker(num);
            this.workerNum += num;
        }
    }

    @Override
    public void removeWorkers(int num) {
        synchronized (jobs) {
            if (num > this.workerNum) {
                throw new IllegalArgumentException("超出工作者数目！");
            }

            int count = 0;
            while (count < num) {
                Worker worker = workers.get(count);
                // 如果移除成功则关闭工作者，工作者将不会继续获取任务执行
                if (workers.remove(worker)) {
                    worker.shutdown();
                    count++;
                }
                this.workerNum -= count;
            }
        }
    }

    @Override
    public int getJobSize() {
        return jobs.size();
    }



    /**
     * <p>
     *      工作者-负责消费客户端提交的任务
     * </p>
     *
     * @Author: Liziba
     * @Date: 2021/6/17 22:41
     */
    class Worker implements Runnable {

        /** 是否工作 */
        private volatile boolean running = Boolean.TRUE;

        @Override
        public void run() {
            while (running) {
                Job job = null;
                synchronized (jobs) {
                    while (jobs.isEmpty()) {
                        try {
                            jobs.wait();
                        } catch (InterruptedException e) {
                            // 如果感应到外部的中断通知，则自己主动中断返回
                            Thread.currentThread().interrupt();
                            return;
                        }
                    }
                    // 取出任务队列的第一个任务
                    job = jobs.removeFirst();
                }
                // 执行任务
                if (job != null) {
                    try {
                        job.run();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        }

        /**
         * 关闭worker,全部关闭意味着线程池关闭
         */
        public void shutdown() {
            running = false;
        }
    }
}

​

测试

package com.lizba.p3.threadpool;

import com.lizba.p2.SleepUtil;

/**
 * <p>
 *      线程池测试
 * </p>
 *
 * @Author: Liziba
 * @Date: 2021/6/17 23:19
 */
public class PoolTest {

    public static void main(String[] args) {

        DefaultThreadPool pool = new DefaultThreadPool();
        // 提交10个任务
        int size = 10;
        for (int i = 0; i < size; i++) {
            Thread job = new Thread(new Runnable() {
                @Override
                public void run() {
                    SleepUtil.sleepSecond(1);
                    System.out.println(Thread.currentThread().getName() + " 执行Job任务");
                }
            });

            pool.execute(job);
        }
    }

}

查看输出：

​

3、总结
从上述代码可以看出，线程池的本质就是使用一个线程安全工作队列来连接工作线程和客户端线程。当客户端调用execute(job)方法提交一个任务时，线程池会向任务列表jobs中添加任务Job，而工作者worker会不断的从jobs中取出一个Job执行，当Job为空时，工作者会进入等待状态。而这个等待/通知就是使用的wait()/notifyAll()来实现的。
​

有经典，有干货，微信搜索【李子捌】关注这个傻瓜式坚持的程序员。