这篇“Java中线程池自定义如何实现”文章的知识点大部分人都不太理解，所以小编给大家总结了以下内容，内容详细，步骤清晰，具有一定的借鉴价值，希望大家阅读完这篇文章能有所收获，下面我们一起来看看这篇“Java中线程池自定义如何实现”文章吧。
线程为什么不能多次调用start方法
从源码可以得知，调用start方法时，程序还会判断当前的线程状态
这里又引申出另一个问题，线程到底有几种状态
年轻的时候背八股文时，只是说五种状态，这五种状态也不知道是哪里来的，不知道有没有人和我一样，当初只是知其然不知其所以然。贴出源码来：
public enum State {
/**
* Thread state for a thread which has not yet started.
*/
NEW, // 新建
/**
* Thread state for a runnable thread. A thread in the runnable
* state is executing in the Java virtual machine but it may
* be waiting for other resources from the operating system
* such as processor.
*/
RUNNABLE, // 运行中
/**
* Thread state for a thread blocked waiting for a monitor lock.
* A thread in the blocked state is waiting for a monitor lock
* to enter a synchronized block/method or
* reenter a synchronized block/method after calling
* {@link Object#wait() Object.wait}.
*/
BLOCKED, // 阻塞
/**
* Thread state for a waiting thread.
* A thread is in the waiting state due to calling one of the
* following methods:
* <ul>
* <li>{@link Object#wait() Object.wait} with no timeout</li>
* <li>{@link #join() Thread.join} with no timeout</li>
* <li>{@link LockSupport#park() LockSupport.park}</li>
* </ul>
*
* <p>A thread in the waiting state is waiting for another thread to
* perform a particular action.
*
* For example, a thread that has called <tt>Object.wait()</tt>
* on an object is waiting for another thread to call
* <tt>Object.notify()</tt> or <tt>Object.notifyAll()</tt> on
* that object. A thread that has called <tt>Thread.join()</tt>
* is waiting for a specified thread to terminate.
*/
WAITING, // 等待
/**
* Thread state for a waiting thread with a specified waiting time.
* A thread is in the timed waiting state due to calling one of
* the following methods with a specified positive waiting time:
* <ul>
* <li>{@link #sleep Thread.sleep}</li>
* <li>{@link Object#wait(long) Object.wait} with timeout</li>
* <li>{@link #join(long) Thread.join} with timeout</li>
* <li>{@link LockSupport#parkNanos LockSupport.parkNanos}</li>
* <li>{@link LockSupport#parkUntil LockSupport.parkUntil}</li>
* </ul>
*/
TIMED_WAITING, // 定时等待
/**
* Thread state for a terminated thread.
* The thread has completed execution.
*/
TERMINATED; // 结束状态
}综上，其实线程的状态有六种：
NEW 新建状态，一般通过Thread thread = new Thread(runable);此时的线程属于新建状态。
RUNABLE 可运行状态，当调用start时，线程进入RUNNABLE状态,该状态其实还包含两个状态，一种是被cpu选中正在运行中，另一种是未被cpu选中，处于就绪状态。
BLOCKED 阻塞状态， 一般可以通过调用sleep()方法来进入阻塞状态，此时线程没有释放锁资源，sleep到期时，继续进入Runable状态
WAITING 等待状态， 一般可以通过调用wait()方法来进入等待状态，此时释放cpu,cpu去干其他事情，需要调用noitfy方法唤醒，唤醒后的线程为RUNABLE状态。
TIMED_WAIRING 定时等待， 一般可以通过wait(long)方法进入定时等待。基本上同WAITING.
TERMINATED 结束状态，RUNCABLE运行正常结束的线程的状态就是TERMINATED
可以看出八股文不能乱背，之前傻呵呵背的八股文很有可能是错误的，比如线程的运行中状态（RUNNING）,其实这个状态根本不存在，RUNABLE状态就已经包含了RUNNNING状态了。
再回到标题的问题，为什么不能多次调用start方法，原因其实源码的注释上已经说明了，
/** * This method is not invoked for the main method thread or "system" * group threads created/set up by the VM. Any new functionality added * to this method in the future may have to also be added to the VM. * * A zero status value corresponds to state "NEW". */
0状态对应的是NEW,也就是说只有新建状态的线程才能调用start方法，其他状态的线程调用就会抛出异常，而一般第二次调用时，线程状态肯定不是new状态了。因此不可以多次调用。
线程池到底是如何复用的
经过多次的反复调试，原理其实很简单，比如以下代码：
public void testThreadPool() {
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 3, 10L, TimeUnit.SECONDS, new ArrayBlockingQueue(3));
threadPoolExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
for (int i=0; i<5; i++) {
threadPoolExecutor.submit(new Runnable() {
@Override
public void run() {
ThreadUtils.doSleep(10000L);
System.out.println(Thread.currentThread().getName() + "--运行");
}
});
}
threadPoolExecutor.shutdown();
}其中循环往threadPoolExecutor中添加的是自定义的业务任务。而真正去运行任务的是线程池中新建的一个线程。因此这里的复用指的是线程池创建出来得这个线程，这个线程并不会销毁，而是循环去队列中获取任务。千万不可理解为线程池复用的线程是使用者自定义的那个业务任务。具体的复用最核心的代码就是下面这段：
while (task != null || (task = getTask()) != null) {
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task);
Throwable thrown = null;
try {
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}这段代码是runworker中的一段代码，线程就是通过循环去获取队列中的任务来达到线程复用的，前台创建多个runable对象，将任务放到runable中，然后将runable放到队列中，线程池创建线程，线程持续循环获取队列中的任务。这就是线程池的实现逻辑。
下面尝试自己去实现一个线程池：该线程只是为了模拟线程池的运行，并未做线程安全的考虑，也未做非核心线程超时回收等功能。
package com.cz.lock.distributed.impl.redis;
import java.util.List;
import java.util.concurrent.*;
/**
* @program: Reids
* @description: 自定义线程池
* @author: Cheng Zhi
* @create: 2023-02-28 09:28
**/
public class JefThreadPoolExecutor extends AbstractExecutorService {
/**
* 使用队列来保存现有的worker
*/
private final BlockingQueue<Worker> workers = new LinkedBlockingQueue<Worker>();
private static int coreThreadCount = 5;
private static int maxThreadCount = 10;
private static int defaultQueueSize = maxThreadCount * 5;
private static BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<Runnable>(defaultQueueSize);
/**
* 默认线程池
*/
JefThreadPoolExecutor() {
this(coreThreadCount, maxThreadCount, blockingQueue);
}
/**
* 可以自定义的线程池
* @param coreThreadCount
* @param maxThreadCount
* @param blockingQueue
*/
JefThreadPoolExecutor(int coreThreadCount, int maxThreadCount, BlockingQueue blockingQueue) {
this.blockingQueue = blockingQueue;
this.coreThreadCount = coreThreadCount;
this.maxThreadCount = maxThreadCount;
}
@Override
public void shutdown() {
}
@Override
public List<Runnable> shutdownNow() {
return null;
}
@Override
public boolean isShutdown() {
return false;
}
@Override
public boolean isTerminated() {
return false;
}
@Override
public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
return false;
}
@Override
public void execute(Runnable command) {
int currentWorkCount = workers.size(); // 当前创建的线程总数
if (currentWorkCount < coreThreadCount) { // 如果当前线程总数小于核心线程数，则新建线程
Worker worker = new Worker(command);
final Thread thread = worker.thread;
thread.start();
addWorker(worker);
return;
}
if (!blockingQueue.offer(command) && currentWorkCount <= maxThreadCount) { // 队列可以正常放入则返回true，如果满了返回false
// 队列如果满了，需要创建新的线程
Worker worker = new Worker(command);
final Thread thread = worker.thread;
thread.start();
addWorker(worker);
return;
} else if (currentWorkCount > maxThreadCount){
System.out.println("线程池满了....没有多余的线程了");
}
}
public void addWorker(Worker worker) {
workers.add(worker);
}
public Runnable getTask() {
Runnable poll = blockingQueue.poll();
return poll;
}
public void runWorker(Worker worker) {
Runnable task = worker.firstTask; // 获取到new Worker时传入的那个任务，并在下面运行
if (task != null) {
task.run();
}
worker.firstTask = null;
// 循环从队列中获取任务处理
while((task = getTask()) != null) {
task.run();
}
}
/**
* 匿名内部类
*/
private class Worker implements Runnable{
volatile int state = 0;
public Runnable firstTask;
final Thread thread;
public Worker(Runnable firstTask) {
this.firstTask = firstTask;
thread = new Thread(this);
}
@Override
public void run() {
runWorker(this);
}
}
}使用方式：
/**
* 使用默认配置
*/
public static void singleThreadPoolExecutor() {
JefThreadPoolExecutor jefThreadPoolExecutor = new JefThreadPoolExecutor();
for (int i=0; i<10; i++) {
jefThreadPoolExecutor.execute(new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "--运行");
}
});
}
}
/**
* 自定义配置
*/
public static void diyThreadPoolExecutor() {
JefThreadPoolExecutor jefThreadPoolExecutor = new JefThreadPoolExecutor(2, 10, new ArrayBlockingQueue(50));
for (int i=0; i<500; i++) {
jefThreadPoolExecutor.execute(new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "--运行");
}
});
}
}