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

目录
  • 前言
  • 线程为什么不能多次调用start方法
  • 线程池到底是如何复用的

前言

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

线程为什么不能多次调用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() + "--运行");
            }
        });
    }
}

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

(0)

相关推荐

  • Java动态线程池插件dynamic-tp集成zookeeper

    目录 前言 配置刷新 Zookeeper配置中心接入扩展实现 总结 前言 dynamic-tp是一个轻量级的动态线程池插件,它是一个基于配置中心的动态线程池,线程池的参数可以通过配置中心配置进行动态的修改,在配置中心的支持上最开始的时候支持Nacos和Apollo,由于笔者公司用的配置中心是Zookeeper,所以就想着扩展支持Zookeeper,在了解源码支持发现dynamic-tp的扩展能力做的很好,提供了扩展接口,只要我开发对应的配置中心模块即可,最终笔者实现了Zookeeper的支持并贡

  • 一文带你弄懂Java中线程池的原理

    目录 为什么要用线程池 线程池的原理 ThreadPoolExecutor提供的构造方法 ThreadPoolExecutor的策略 线程池主要的任务处理流程 ThreadPoolExecutor如何做到线程复用的 四种常见的线程池 newCachedThreadPool newFixedThreadPool newSingleThreadExecutor newScheduledThreadPool 小结 在工作中,我们经常使用线程池,但是你真的了解线程池的原理吗?同时,线程池工作原理和底层实

  • Java实现手写线程池实例并测试详解

    前言 在之前的文章中介绍过线程池的核心原理,在一次面试中面试官让手写线程池,这块知识忘记的差不多了,因此本篇文章做一个回顾. 希望能够加深自己的印象以及帮助到其他的小伙伴儿们 在线程池核心原理篇介绍过线程池的核心原理,今天来模拟线程池和工作队列的流程,以及编写代码和测试类进行测试.下面附下之前线程池的核心流程: 在线程池核心原理的源码中,涉及到了一系列的流程,包括线程池队列数量是否已满,运用什么样的拒绝策略等.在我们手写线程池的代码中,不需要考虑那么多因素,只需要模拟简单的情景和过程,因此整体来

  • Java动态线程池插件dynamic-tp集成过程浅析

    目录 前言 快速开始 监控数据和第三方平台对接 logging方式 MicroMeter方式 总结 前言 dynamic-tp是一个轻量级的动态线程池插件,它是一个基于配置中心的动态线程池,线程池的参数可以通过配置中心配置进行动态的修改,目前支持的配置中心有Apollo,Nacos和Zookeeper,同时dynamic-tp支持线程池的监控和报警,具体特性如下: 基于Spring框架,现只支持SpringBoot项目使用,轻量级,引入 starter 即可使用 基于配置中心实现线程池参数动态调

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

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

  • Java线程池Executor用法详解

    目录 线程池类图 线程池的好处 new Thread的弊端 线程池核心类-ThreadPoolExecutor 使用Executors创建线程池 Executors.newCachedThreadPool Executors.newSingleThreadExecutor Executors.newFixedThreadPool Executors.newScheduledThreadPool 总结 如何定义线程池参数 线程池类图 我们最常使用的Executors实现创建线程池使用线程主要是用上

  • java 打造阻塞式线程池的实例详解

    java 打造阻塞式线程池的实例详解 原来以为tiger已经自带了这种线程池,就是在任务数量超出时能够阻塞住投放任务的线程,主要想用在JMS消息监听. 开始做法: 在ThreadPoolExcecutor中代入new ArrayBlockingQueue(MAX_TASK). 在任务超出时报错:RejectedExecutionException. 后来不用execute方法加入任务,直接getQueue().add(task), 利用其阻塞特性.但是发现阻塞好用了,但是任务没有被处理.一看Qu

  • 基于Java中的数值和集合详解

    数组array和集合的区别: (1) 数值是大小固定的,同一数组只能存放一样的数据. (2) java集合可以存放不固定的一组数据 (3) 若程序事不知道究竟需要多少对象,需要在空间不足时自动扩增容量,则需要使用容器类库,array不适用 数组转换为集合: Arrays.asList(数组) 示例: int[] arr = {1,3,4,6,6}; Arrays.asList(arr); for(int i=0;i<arr.length;i++){ System.out.println(arr[

  • java中Spring Security的实例详解

    java中Spring Security的实例详解 spring security是一个多方面的安全认证框架,提供了基于JavaEE规范的完整的安全认证解决方案.并且可以很好与目前主流的认证框架(如CAS,中央授权系统)集成.使用spring security的初衷是解决不同用户登录不同应用程序的权限问题,说到权限包括两部分:认证和授权.认证是告诉系统你是谁,授权是指知道你是谁后是否有权限访问系统(授权后一般会在服务端创建一个token,之后用这个token进行后续行为的交互). spring

  • 多用多学之Java中的Set,List,Map详解

    很长时间以来一直代码中用的比较多的数据列表主要是List,而且都是ArrayList,感觉有这个玩意就够了.ArrayList是用于实现动态数组的包装工具类,这样写代码的时候就可以拉进拉出,迭代遍历,蛮方便的. 也不知道从什么时候开始慢慢的代码中就经常会出现HashMap和HashSet之类的工具类.应该说HashMap比较多一些,而且还是面试经典题,平时也会多看看.开始用的时候简单理解就是个键值对应表,使用键来找数据比较方便.随后深入了解后发现 这玩意还有点小奥秘,特别是新版本的JDK对Has

  • Java中CountDownLatch进行多线程同步详解及实例代码

    Java中CountDownLatch进行多线程同步详解 CountDownLatch介绍 在前面的Java学习笔记中,总结了Java中进行多线程同步的几个方法: 1.synchronized关键字进行同步. 2.Lock锁接口及其实现类ReentrantLock.ReadWriteLock锁实现同步. 3.信号量Semaphore实现同步. 其中,synchronized关键字和Lock锁解决的是多个线程对同一资源的并发访问问题.信号量Semaphore解决的是多副本资源的共享访问问题. 今天

  • java 中死锁问题的实例详解

    java 中死锁问题的实例详解 先看代码在做解释 public class DeadLock implements Runnable{ String a; String b; boolean flag; public DeadLock(String a,String b,boolean flag){ this.a=a; this.b=b; this.flag=flag; } public void run(){ if(flag){ // while(true){ synchronized(a){

  • Java 中synchronize函数的实例详解

    Java 中synchronize函数的实例详解 java中的一个类的成员函数若用synchronized来修饰,则对应同一个对象,多个线程像调用这个对象的这个同步函数时必须等到上一个线程调用完才能由下一个线程调用. 那么如果一个类同时有两个成员函数是由synchronized修饰如代码所示,对与同一个对象,是否可以在两个线程运行时,一个调用funcA,同时另一个调用funcB? Mysyn是这样一个类,如果我有两个线程,一个在run方法中先运行funcA再运行funcB,另一个线程在run方法

  • Java中Lambda表达式的使用详解

    目录 理解函数式接口以及 Lambda表达式的发展过程 Lambda表达式及语法 一起来看看具体的使用 你需要注意什么 Lambda的实际运用 1.对集合排序 2.遍历集合 3.遍历集合(带条件) 4.代替 Runnable,开启一个线程 理解函数式接口以及 Lambda表达式的发展过程 任何接口,只包含唯一一个抽象方法,就是函数式接口 /** * lambdab表达式的发展 */ public class TestLambda1 { //3.静态内部类 static class Like2 i

随机推荐