线程池设计

使用线程池的场合

• 单个任务处理时间短
• 将需处理的任务数量大

使用Java线程池好处

1、使用new Thread()创建线程的弊端：

• 每次通过new Thread()创建对象性能不佳。
• 线程缺乏统一管理，可能无限制新建线程，相互之间竞争，及可能占用过多系统资源导致死机或oom。
• 缺乏更多功能，如定时执行、定期执行、线程中断。
  2、使用Java线程池的好处：
• 重用存在的线程，减少对象创建、消亡的开销，提升性能。
• 可有效控制最大并发线程数，提高系统资源的使用率，同时避免过多资源竞争，避免堵塞。
• 提供定时执行、定期执行、单线程、并发数控制等功能。

一般一个简单线程池至少包含下列组成部分。

• 线程池管理器（CustomThreadPool）:用于创建并管理线程池
• 工作线程（CustomWorker）: 线程池中线程
• 任务接口(基类)（CustomTask）:每个任务必须实现的接口（或继承的基类），以供工作线程调度任务的执行。
• 任务队列:用于存放没有处理的任务。提供一种缓冲机制。
  详细代码

线程池管理器至少有下列功能：创建线程池，销毁线程池，添加新任务
线程池的代码如下:

package com.tao.threadpool;

import lombok.extern.slf4j.Slf4j;

import java.util.Optional;
import java.util.Vector;
import java.util.concurrent.BlockingQueue;

/**
 * @author: Penger
 * @time: 2019/7/23
 * @description: <p>
 * 自定义线程池简单实现：
 * 1. 线程池大小
 * 2. 创建、销毁线程池
 * 3. 工作线程
 * 4. 任务队列
 * </p>
 **/
@Slf4j
class CustomThreadPool {
    /**
     * 状态，是否在运行
     */
    private volatile Boolean running = true;
    /**
     * 任务缓存队列
     */
    private final BlockingQueue<CustomTask> taskQueue;
    /**
     * worker的执行需要thread
     */
    private Vector<CustomWorker> workers = new Vector<>();
    /**
     * 最大线程数
     */
    private int maxPoolSize;
    /**
     * 当前线程数
     */
    private int workSize;

    CustomThreadPool(int maxPoolSize, BlockingQueue<CustomTask> taskQueue) {
        this.maxPoolSize = maxPoolSize;
        this.workSize = 0;
        this.taskQueue = taskQueue;
    }

    void execute(CustomTask task) {
        Optional.ofNullable(task).orElseThrow(NullPointerException::new);
        if (workSize < maxPoolSize) {
            addThread(task);
        } else if (!taskQueue.offer(task)) {
            rejectTask(task);
        }
    }

    /**
     * 添加线程执行start
     *
     * @param task task
     */
    private void addThread(CustomTask task) {
        //当前线程数+1
        workSize++;
        //创建新的Worker
        CustomWorker worker = new CustomWorker("Thread-" + workSize, task, taskQueue, running);
        workers.add(worker);
        //执行thread
        worker.start();
    }

    /**
     * 停止所有线程
     */
    void shutdown() {
        running = false;
        if (!workers.isEmpty()) {
            for (CustomWorker worker : workers) {
//                worker.setRunning(false);
                worker.interruptSelf();
            }
        }
        Thread.currentThread().interrupt();
    }

    private void rejectTask(CustomTask task) {
        log.info("blocking queue has no space, reject task{}", task.getTaskId());
    }
}

工作线程可以继承thread，能够获取到任务队列，实现Thread的run方法，在run里面取任务执行,同时提供工作线程的销毁方法

 package com.tao.threadpool;

import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.BlockingQueue;

/**
 * @author: Penger
 * @time: 2019/7/23
 * @description: <p>
 * 自定义工作线程,优先执行直接赋予的任务，执行完毕之后再从队列获取任务
 * </p>
 **/
@EqualsAndHashCode(callSuper = true)
@Slf4j
@Data
public class CustomWorker extends Thread {
    private CustomTask task;
    private final BlockingQueue<CustomTask> taskQueue;
    private Boolean running;

    CustomWorker(String name, CustomTask task, BlockingQueue<CustomTask> taskQueue, Boolean running) {
        super(name);
        this.task = task;
        this.taskQueue = taskQueue;
        this.running = running;
    }

    @Override
    public void run() {
        while (running) {
            if (task != null) {
                task.taskRun();
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            try {
                task = taskQueue.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.info("queue has [{}] elements", taskQueue.size());
        }
    }

    void interruptSelf() {
        this.interrupt();
    }
}

任务接口(基类)，定义线程池具体执行何种任务，通过taskRun()开始任务执行

package com.tao.threadpool;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

/**
 * @author: Penger
 * @time: 2019/7/23
 * @description: <p>
 * 自定义任务
 * </p>
 **/
@Slf4j
@Data
class CustomTask {
    private int taskId;

    CustomTask(int taskId) {
        this.taskId = taskId;
    }

    void taskRun() {
        log.info("custom task-{} running...", taskId);
    }
}