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

2023-03-01 11:03:28 线程自定义详解

前言

最初使用线程池的时候，网上的文章告诉我说线程池可以线程复用，提高线程的创建效率。从此我的脑海中便为线程池打上了一个标签——线程池可以做到线程的复用。但是我总以为线程的复用是指在创建出来的线程可以多次的更换run（）方法的内容，来达到线程复用的目的，于是我尝试了一下.同一个线程调用多次，然后使run的内容不一样，但是我发现我错了，一个线程第一次运行是没问题的，当再次调用start方法是会抛出异常（java.lang.IllegalThreadStateException）。

线程为什么不能多次调用start方法

从源码可以得知，调用start方法时，程序还会判断当前的线程状态

这里又引申出另一个问题，线程到底有几种状态

年轻的时候背八股文时，只是说五种状态，这五种状态也不知道是哪里来的，不知道有没有人和我一样，当初只是知其然不知其所以然。贴出源码来：

public enum State {
    
    NEW, // 新建

    
    RUNNABLE, // 运行中

    
    BLOCKED, // 阻塞

    
    WaiTING, // 等待

    
    TIMED_WAITING, // 定时等待

    
    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方法，原因其实源码的注释上已经说明了，

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.*;


public class JefThreadPoolExecutor extends AbstractExecutorService {

    
    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);
    }
    
    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() + "--运行");
            }
        });
    }
}

以上就是Java中线程池自定义实现详解的详细内容，更多关于Java线程池的资料请关注其它相关文章！

相关文章

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

目录

前言

线程为什么不能多次调用start方法

线程池到底是如何复用的