Java中线程池自定义实现详解

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: * 
 *   
{@link Object#wait() Object.wait} with no timeout
 *   
{@link #join() Thread.join} with no timeout
 *   
{@link LockSupport#park() LockSupport.park}
 * 
 * * 
A thread in the waiting state is waiting for another thread to * perform a particular action. * * For example, a thread that has called Object.wait() * on an object is waiting for another thread to call * Object.notify() or Object.notifyAll() on * that object. A thread that has called Thread.join() * 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: * 
 *   
{@link #sleep Thread.sleep}
 *   
{@link Object#wait(long) Object.wait} with timeout
 *   
{@link #join(long) Thread.join} with timeout
 *   
{@link LockSupport#parkNanos LockSupport.parkNanos}
 *   
{@link LockSupport#parkUntil LockSupport.parkUntil}
 * 
 */ TIMED_WAITING, // 定时等待 /** * Thread state for a terminated thread. * The thread has completed execution. */ TERMINATED; // 结束状态 }

/** * 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". */

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(); } 

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(); } }

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 workers = new LinkedBlockingQueue(); private static int coreThreadCount = 5; private static int maxThreadCount = 10; private static int defaultQueueSize = maxThreadCount * 5; private static BlockingQueue blockingQueue = new ArrayBlockingQueue(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 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  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() + "--运行"); } }); } }