java线程池的使用(jdk1.8)

Posted 2022-02-20 alive

今天是2022-02-19，周六。今天我们来聊聊java的线程池。大家听到线程池，第一反应是联想到线程。那么何为线程池？跟线程有啥区别？以及线程池、线程都是怎么使用？带着这些疑问，看完这篇大家有几本的了解。

 

一、线程池是什么？

线程池就是可以创建固定线程数量、最大线程数、等待队列数、每一个线程的执行时间、线程的名称等参数的线程。大家可以理解成，线程池就是多个线程组成，但是跟线程又有区别。线程是单一且需要时就创建，执行完任务就销毁，而线程池就不会，需要就取一个创建好的线程，用完就放回去。

 

二、创建线程池有哪些方式？(使用Executors顶层容器静态类实现创建)

  1、Executors.newCachedThreadPool(); //创建一个缓冲池，缓冲池容量大小为Integer.MAX_VALUE

  2、Executors.newSingleThreadExecutor(); //创建容量为1的缓冲池

  3、Executory.newFixedThreadPool(int); //创建固定容量大小的线程池

 

三、使用ThreadPoolExecutor创建线程池（参数可以定义配置到配置到配置文件）

下面直接上代码：

import cn.hutool.json.JSONUtil;
import com.huawei.wps.config.properties.AsyncConvertProgressThreadPoolProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.ThreadPoolExecutor;

@Configuration
@EnableAsync //这个注解是因为我后面使用的是异步线程池，所以加了这个开启异步
@Slf4j //日志
public class AsyncConvertProgressThreadPoolTaskConfig 
    /**
     * 默认情况下，在创建了线程池后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，
     * 当线程池中的线程数目达到corePoolSize后，就会把到达的任务放到缓存队列当中；
     * 当队列满了，就继续创建线程，当线程数量大于等于maxPoolSize后，开始使用拒绝策略拒绝
     */

    @Autowired
    private AsyncThreadPoolProperties asyncThreadPoolProperties;

    //配置线程池
    @Bean("AsycnTaskExecutor")
    public ThreadPoolTaskExecutor convertProgressTaskExecutor() 
        log.info("加载异步线程池:", JSONUtil.toJsonStr(asyncThreadPoolProperties)); //加载参数
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); //使用ThreadPoolTaskExector线程池类
        executor.setCorePoolSize(asyncThreadPoolProperties.getCorePoolSize()); //配置核心线程数
        executor.setMaxPoolSize(asyncThreadPoolProperties.getMaxPoolSize()); //配置最大线程数
        executor.setQueueCapacity(asyncThreadPoolProperties.getQueueCapacity());//配置等待线程池数的容量
        executor.setKeepAliveSeconds(asyncThreadPoolProperties.getKeepAliveTime()); //
        executor.setThreadNamePrefix(asyncThreadPoolProperties.getThreadNamePrefix()); //

        // 线程池对拒绝任务的处理策略
        // CallerRunsPolicy：由调用线程（提交任务的线程）处理该任务
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); //配置拒绝策略
        // 初始化
        executor.initialize();
        return executor;
    

 

上面是配置，那么如何使用这个线程池？

 @Async("AsyncTaskExecutor")//该注解是实现下面的方法异步，至于：AsyncTaskExecutory是因为绑定了上面的Bean实例，也就是使用线程池。完整的解释是：使用异步线程池来完成方法：asyncTask的业务处理
 public void asyncTask(Integer paramater,String phone) 

上面是参数配置，以及异步线程池的使用。目前我在实际工作中，文件进行处理（分片上传，分片下载），使用该异步线程池，都能满足业务的需要（这里不能说是最优，毕竟其他小伙伴肯定其他更好的方案）。

 

四、扒拉原理（待讲解）

 大家看到了异步线程池主要是，那么我们来看看里面有哪些逻辑代码：ThreadPoolTaskExecutor

ThreadPoolTaskExecutor: 原来是继承了ExecutorConfigurationSupport和实现AnyncListanbleTaskExecutory、SchedulingTaskExecutor

 

 

ExecutorConfigurationSupport.class如下：

public abstract class ExecutorConfigurationSupport extends CustomizableThreadFactory implements BeanNameAware, InitializingBean, DisposableBean 
    protected final Log logger = LogFactory.getLog(this.getClass());
    private ThreadFactory threadFactory = this;
    private boolean threadNamePrefixSet = false;
    private RejectedExecutionHandler rejectedExecutionHandler = new AbortPolicy();  //给了默认拒绝策略
    private boolean waitForTasksToCompleteOnShutdown = false;
    private long awaitTerminationMillis = 0L;
    @Nullable
    private String beanName;
    @Nullable
    private ExecutorService executor;

    public ExecutorConfigurationSupport() 
    
    
    //省了其它get、set代码

 

AsyncListenableTaskExecutor.class如下：

public interface AsyncListenableTaskExecutor extends AsyncTaskExecutor 

    /**
     * Submit a @code Runnable task for execution, receiving a @code ListenableFuture
     * representing that task. The Future will return a @code null result upon completion.
     * @param task the @code Runnable to execute (never @code null)
     * @return a @code ListenableFuture representing pending completion of the task
     * @throws TaskRejectedException if the given task was not accepted
     */
    ListenableFuture<?> submitListenable(Runnable task);

    /**
     * Submit a @code Callable task for execution, receiving a @code ListenableFuture
     * representing that task. The Future will return the Callable\'s result upon
     * completion.
     * @param task the @code Callable to execute (never @code null)
     * @return a @code ListenableFuture representing pending completion of the task
     * @throws TaskRejectedException if the given task was not accepted
     */
    <T> ListenableFuture<T> submitListenable(Callable<T> task);

 

接着：AsyncTaskExecutor接口如下：你会发现只有execute、sumbit抽象方法 

public interface AsyncTaskExecutor extends TaskExecutor 

    /** Constant that indicates immediate execution. */
    long TIMEOUT_IMMEDIATE = 0;

    /** Constant that indicates no time limit. */
    long TIMEOUT_INDEFINITE = Long.MAX_VALUE;


    /**
     * Execute the given @code task.
     * @param task the @code Runnable to execute (never @code null)
     * @param startTimeout the time duration (milliseconds) within which the task is
     * supposed to start. This is intended as a hint to the executor, allowing for
     * preferred handling of immediate tasks. Typical values are @link #TIMEOUT_IMMEDIATE
     * or @link #TIMEOUT_INDEFINITE (the default as used by @link #execute(Runnable)).
     * @throws TaskTimeoutException in case of the task being rejected because
     * of the timeout (i.e. it cannot be started in time)
     * @throws TaskRejectedException if the given task was not accepted
     */
    void execute(Runnable task, long startTimeout);

    /**
     * Submit a Runnable task for execution, receiving a Future representing that task.
     * The Future will return a @code null result upon completion.
     * @param task the @code Runnable to execute (never @code null)
     * @return a Future representing pending completion of the task
     * @throws TaskRejectedException if the given task was not accepted
     * @since 3.0
     */
    Future<?> submit(Runnable task);

    /**
     * Submit a Callable task for execution, receiving a Future representing that task.
     * The Future will return the Callable\'s result upon completion.
     * @param task the @code Callable to execute (never @code null)
     * @return a Future representing pending completion of the task
     * @throws TaskRejectedException if the given task was not accepted
     * @since 3.0
     */
    <T> Future<T> submit(Callable<T> task);

 

接着继续看：TaskExecutor接口：只有重写了Executor接口 的execute抽象方法

@FunctionalInterface
public interface TaskExecutor extends Executor 

    /**
     * Execute the given @code task.
     * <p>The call might return immediately if the implementation uses
     * an asynchronous execution strategy, or might block in the case
     * of synchronous execution.
     * @param task the @code Runnable to execute (never @code null)
     * @throws TaskRejectedException if the given task was not accepted
     */
    @Override
    void execute(Runnable task);

 

那么我们继续看：Executor接口————————你会发现，原来异步线程池，主要是还是使用了顶层容器：Executor接口

public interface Executor 

    /**
     * Executes the given command at some time in the future.  The command
     * may execute in a new thread, in a pooled thread, or in the calling
     * thread, at the discretion of the @code Executor implementation.
     *
     * @param command the runnable task
     * @throws RejectedExecutionException if this task cannot be
     * accepted for execution
     * @throws NullPointerException if command is null
     */
    void execute(Runnable command);

 

那么，我们一起看看execute方法，具体逻辑：ThreadPoolTaskExecutor--》execute，看源码如下：

    public void execute(Runnable task) 
        ThreadPoolExecutor executor = this.getThreadPoolExecutor();

        try 
            executor.execute(task); //我们继续进入该方法，看看里面的逻辑是什么
         catch (RejectedExecutionException var4)  //这里是拒绝策略
            throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, var4);
        
    

我们继续进入该方法，看看里面的逻辑是什么：executor.execute(task); 

 /**
     * Executes the given task sometime in the future.  The task
     * may execute in a new thread or in an existing pooled thread.
     *
     * If the task cannot be submitted for execution, either because this
     * executor has been shutdown or because its capacity has been reached,
     * the task is handled by the current @code RejectedExecutionHandler.
     *
     * @param command the task to execute
     * @throws RejectedExecutionException at discretion of
     *         @code RejectedExecutionHandler, if the task
     *         cannot be accepted for execution
     * @throws NullPointerException if @code command is null
     */
    public void execute(Runnable command) 
       //这里不用讲解了吧。。。。大家都看得懂
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn\'t, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
         */
        int c = ctl.get(); //上面这个变量的定义是：AtomicInteger ctl=new AtomicInteger(ctlOf(RUNNING,0)) //得到工作队列数
        if (workerCountOf(c) < corePoolSize)  //
            if (addWorker(command, true)) //添加进工作队列
                return;
            c = ctl.get();
        
        if (isRunning(c) && workQueue.offer(command))  //判断如果当前线程任务正在运行，那么把线程放入工作队列中，并且成功，进入if方法
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command)) //再次判断该线程不运行了，并且从工作队列中移除成功
                reject(command); //拒绝该线程
            else if (workerCountOf(recheck) == 0) //如果正在工作的线程数量为0,进入下面的addWorker
                addWorker(null, false); //null参数的意思是，没有线程可以加入工作队列
        
        else if (!addWorker(command, false))
            reject(command);