Young87

SmartCat's Blog

So happy to code my life!

游戏开发交流QQ群号60398951

当前位置：首页 >跨站数据测试

线程池ThreadPoolExecutor使用

日期： 2020-12-14    分类： 跨站数据测试 567次阅读

ThreadPoolExecutor线程池使用

粗略使用ThreadPoolExecutor

ThreadPoolExecutor参数

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)

corePoolSize-核心线程池大小

  • 当提交一个任务到线程池时，线程池会创建一个线程来执行任务，即使其他空闲的基本线程能够执行新任务也会创建线程，等到需要执行的任务数大于线程池基本大小时就不再创建

maximumPoolSize-最大线程池大小

  • 如果队列满了，并且已创建的线程数小于最大线程数，则线程池会再创建新的线程执行任务
  • PS 如果使用了无界的任务队列这个参数就没什么效果

keepAliveTime-线程最大空闲时间

线程池的工作线程空闲后，保持存活的时间

  • 如果任务很多，并且每个任务执行的时间比较短，可以调大这个时间，提高线程的利用率
  • 此参数对核心线程数的线程无作用，仅对超过核心线程大小且不超过最大线程池时创建对线程有效

TimeUnit-时间单位

指定线程最大空闲时间的单位

  • NANOSECONDS: 毫微秒 十亿分之一秒【1微秒/1000】
  • MICROSECONDS: 微秒 一百万分之一秒【1毫秒/1000】
  • MILLISECONDS: 毫秒 千分之一秒
  • SECONDS: 秒
  • MINUTES: 分钟
  • HOURS: 小时
  • DAYS: 天

blockingQueue-线程等待队列

用于保存等待执行的任务的阻塞队列

  • 可以选择以下几个阻塞队列：
  • ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，此队列按 FIFO（先进先出）原则对元素进行排序
  • LinkedBlockingQueue：一个基于链表结构的阻塞队列，此队列按FIFO （先进先出） 排序元素，吞吐量通常要高于ArrayBlockingQueue
  • SynchronousQueue：一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue
  • PriorityBlockingQueue：一个具有优先级的无限阻塞队列

threadFactory-线程工厂

可以通过线程工厂给每个创建出来的线程做些更有意义的事情,比如设置线程线程名称和优先级等

  • 目前可以通过三个工具包创建线程工厂
  • Spring 框架提供的 CustomizableThreadFactory
  • Google guava 工具类 提供的 ThreadFactoryBuilder
  • Apache commons-lang3 提供的 BasicThreadFactory

RejectedExecutionHandler-饱和策略

当队列和线程池都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务
这个策略默认情况下是AbortPolicy，表示无法处理新任务时抛出异常
PS 当使用DiscardPolicy时，如果使用线程是Future接收线程结果，如果get时不设置超时时间，则无限阻塞主线程（注意！）

  • 可以选择以下几个策略：
  • AbortPolicy：直接抛出异常
  • CallerRunsPolicy：只用调用者所在线程来运行任务
  • DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务
  • DiscardPolicy：不处理，丢弃掉
  • 自定义Reject： 实现RejectedExecutionHandler

预定义线程池

固定线程池

 	/**
     * 预定义线程池 - 固定线程池
     * 队列默认使用：LinkedBlockingQueue(一个基于链表结构的阻塞队列，此队列按FIFO （先进先出） 排序元素，吞吐量通常要高于ArrayBlockingQueue)
     * PS 适用场景：可用于Web服务瞬时削峰，但需注意长时间持续高峰情况造成的队列阻塞。
     * @param nThreads 线程数
     * @param threadFactory 线程工厂（可设置线程名称）
     */
    public static ExecutorService buildFixedThreadPool(int nThreads, ThreadFactory threadFactory){
        return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), threadFactory);
    }

无限线程池

	/**
     * 预定义线程池 - 无限线程池
     * 线程数最大为Integer.MAX_VALUE
     * 队列默认使用：SynchronousQueue(个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue)
     * 线程默认空闲时间：60s
     * PS 适用场景：快速处理大量耗时较短的任务，如Netty的NIO接受请求时，可使用CachedThreadPool
     * @param threadFactory 线程工厂（可设置线程名称）
     */
    public static ExecutorService buildCachedThreadPool(ThreadFactory threadFactory){
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<>(), threadFactory);
    }

线程池使用示例

  • 线程任务
package com.panda.service.test.task.thread;

import com.alibaba.fastjson.JSONObject;
import com.panda.service.test.task.vo.OrderCheckInfo;
import com.panda.service.test.task.vo.OrderCheckResponse;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Callable;

/**
 * 订单检查异步线程处理
 * @author w
 * @date 2020-12-09
 */
@Slf4j
public class OrderCheckThread implements Callable<OrderCheckResponse> {

    /** 订单信息 **/
    private OrderCheckInfo orderCheckInfo;

    public OrderCheckThread(OrderCheckInfo orderCheckInfo){
        this.orderCheckInfo = orderCheckInfo;
    }

    @Override
    public OrderCheckResponse call() {
        try{
            log.info("开始执行订单检查:{}", JSONObject.toJSONString(orderCheckInfo));
            return new OrderCheckResponse(true, "交易成功", orderCheckInfo.getOrderId());
        }catch (Exception e){
            log.error(e.getMessage(),e);
            return new OrderCheckResponse(false, "交易异常" + e.getMessage(), orderCheckInfo.getOrderId());
        }
    }
}
  • 构建线程池并执行线程
// 采用有界队列，长度10，LinkedBlockingQueue
// 饱和策略采用DiscardPolicy，剩下的任务丢掉不执行
ExecutorService pool = new ThreadPoolExecutor(
                    10,
                    20,
                    0L,
                    TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<>(10),
                    new ThreadPoolExecutor.AbortPolicy());


// 创建任务
OrderCheckThread thread = new OrderCheckThread(new OrderCheckInfo(System.currentTimeMillis()))

// 提交任务
Future<OrderCheckResponse> future = pool.submit(thread);

// 获取执行结果，此处会阻塞主线程等待子线程执行完成
Future<OrderCheckResponse> result = future.get();

// 关闭线程池
pool.shutdown();

完整案例

<设计思路>

  • 封装线程工厂，提供相关API，让调用者只需关系业务处理，无需了解线程执行过程;
  • 增加统一性，考虑后续统一管理

<示例>

线程配置信息

package com.panda.service.pool.config;

import java.util.concurrent.*;

/**
 * ThreadFactory Config
 * @author w
 * @date 2020-12-08
 */
public class ThreadFactoryConfig {
    /**
     * 核心线程池大小
     * 当提交一个任务到线程池时，线程池会创建一个线程来执行任务，即使其他空闲的基本线程能够执行新任务也会创建线程，等到需要执行的任务数大于线程池基本大小时就不再创建
     */
    private int corePoolSize;

    /**
     * 最大线程池大小
     * 如果队列满了，并且已创建的线程数小于最大线程数，则线程池会再创建新的线程执行任务
     * PS 如果使用了无界的任务队列这个参数就没什么效果
     */
    private int maximumPoolSize;

    /**
     * 线程最大空闲时间（线程池的工作线程空闲后，保持存活的时间）
     * 如果任务很多，并且每个任务执行的时间比较短，可以调大这个时间，提高线程的利用率
     */
    private long keepAliveTime;

    /**
     * 时间单位
     */
    private TimeUnit timeUnit;


    /**
     * 线程等待队列（用于保存等待执行的任务的阻塞队列）
     * 可以选择以下几个阻塞队列：
     * - ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，此队列按 FIFO（先进先出）原则对元素进行排序
     * - LinkedBlockingQueue：一个基于链表结构的阻塞队列，此队列按FIFO （先进先出） 排序元素，吞吐量通常要高于ArrayBlockingQueue
     * - SynchronousQueue：一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue
     * - PriorityBlockingQueue：一个具有优先级的无限阻塞队列
     */
    private BlockingQueue<Runnable> blockingQueue;

    /**
     * 用于设置创建线程的工厂
     * 可以通过线程工厂给每个创建出来的线程做些更有意义的事情,比如设置线程线程名称和优先级等
     */
    private ThreadFactory threadFactory;

    /**
     * 饱和策略
     * 当队列和线程池都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。
     * 这个策略默认情况下是AbortPolicy，表示无法处理新任务时抛出异常
     * 可以选择以下几个策略：
     *  - AbortPolicy：直接抛出异常
     *  - CallerRunsPolicy：只用调用者所在线程来运行任务
     *  - DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务
     *  - DiscardPolicy：不处理，丢弃掉
     *  - 自定义Reject： 实现RejectedExecutionHandler
     */
    private RejectedExecutionHandler rejectedExecutionHandler;

    public int getCorePoolSize() {
        return corePoolSize;
    }

    public void setCorePoolSize(int corePoolSize) {
        this.corePoolSize = corePoolSize;
    }

    public int getMaximumPoolSize() {
        return maximumPoolSize;
    }

    public void setMaximumPoolSize(int maximumPoolSize) {
        this.maximumPoolSize = maximumPoolSize;
    }

    public long getKeepAliveTime() {
        return keepAliveTime;
    }

    public void setKeepAliveTime(long keepAliveTime) {
        this.keepAliveTime = keepAliveTime;
    }

    public TimeUnit getTimeUnit() {
        return timeUnit;
    }

    public void setTimeUnit(TimeUnit timeUnit) {
        this.timeUnit = timeUnit;
    }

    public BlockingQueue<Runnable> getBlockingQueue() {
        return blockingQueue;
    }

    public void setBlockingQueue(BlockingQueue<Runnable> blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    public ThreadFactory getThreadFactory() {
        return threadFactory;
    }

    public void setThreadFactory(ThreadFactory threadFactory) {
        this.threadFactory = threadFactory;
    }

    public RejectedExecutionHandler getRejectedExecutionHandler() {
        return null != rejectedExecutionHandler ? rejectedExecutionHandler : new ThreadPoolExecutor.AbortPolicy();
    }

    public void setRejectedExecutionHandler(RejectedExecutionHandler rejectedExecutionHandler) {
        this.rejectedExecutionHandler = rejectedExecutionHandler;
    }
}

线程内置工厂

package com.panda.service.pool.factory;

import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import java.util.concurrent.ThreadFactory;

public class NameThreadFactoryBuilder {
    /**
     * 创建一个指定线程名称的线程工厂
     * @param threadName 线程名称
     * @return ThreadFactory
     */
    public static ThreadFactory buildThreadFactory(String threadName){
        return new com.google.common.util.concurrent.ThreadFactoryBuilder().setNameFormat(threadName).build();
    }

    /**
     * 创建一个指定线程名称的线程工厂
     * @param threadName 线程名称
     * @return ThreadFactory
     */
    public static ThreadFactory buildCustomizableThreadFactory(String threadName){
        return new CustomizableThreadFactory(threadName);
    }

    /**
     * 创建一个指定线程名称的线程工厂
     * @param threadName 线程名称
     * @return ThreadFactory
     */
    public static ThreadFactory buildBasicThreadFactory(String threadName){
        return new BasicThreadFactory.Builder().namingPattern(threadName).build();
    }
}

构建线程池

package com.panda.service.pool.factory;

import com.panda.service.pool.config.ThreadFactoryConfig;

import java.util.Optional;
import java.util.concurrent.*;

/**
 * 构建线程池
 * @author w
 * @date 2020-12-09
 */
public class ThreadFactoryBuilder {
    /**
     * 预定义线程池 - 固定线程池
     * 队列默认使用：LinkedBlockingQueue(一个基于链表结构的阻塞队列，此队列按FIFO （先进先出） 排序元素，吞吐量通常要高于ArrayBlockingQueue)
     * PS 适用场景：可用于Web服务瞬时削峰，但需注意长时间持续高峰情况造成的队列阻塞。
     * @param nThreads 线程数
     * @param threadFactory 线程工厂（可设置线程名称）
     */
    public static ExecutorService buildFixedThreadPool(int nThreads, ThreadFactory threadFactory){
        return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), threadFactory);
    }

    /**
     * 预定义线程池 - 无限线程池
     * 线程数最大为Integer.MAX_VALUE
     * 队列默认使用：SynchronousQueue(个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue)
     * 线程默认空闲时间：60s
     * PS 适用场景：快速处理大量耗时较短的任务，如Netty的NIO接受请求时，可使用CachedThreadPool
     * @param threadFactory 线程工厂（可设置线程名称）
     */
    public static ExecutorService buildCachedThreadPool(ThreadFactory threadFactory){
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<>(), threadFactory);
    }

    /**
     * 构建自定义线程池
     * @param config 线程池配置
     */
    public static ExecutorService buildThreadPool(ThreadFactoryConfig config){
        // 检查
        Optional.ofNullable(config.getBlockingQueue()).orElseThrow(() -> new RuntimeException("BlockingQueue is null !"));
        Optional.ofNullable(config.getTimeUnit()).orElseThrow(() -> new RuntimeException("TimeUnit is null !"));

        // 构建线程池
        ExecutorService pool;
        if(null == config.getThreadFactory()){
            pool = new ThreadPoolExecutor(
                    config.getCorePoolSize(),
                    config.getMaximumPoolSize(),
                    config.getKeepAliveTime(),
                    config.getTimeUnit(),
                    config.getBlockingQueue(),
                    config.getRejectedExecutionHandler());
        }else{
            pool = new ThreadPoolExecutor(
                    config.getCorePoolSize(),
                    config.getMaximumPoolSize(),
                    config.getKeepAliveTime(),
                    config.getTimeUnit(),
                    config.getBlockingQueue(),
                    config.getThreadFactory(),
                    config.getRejectedExecutionHandler());
        }
        return pool;
    }

}

自定义饱和策略

package com.panda.service.pool.reject;

import java.util.concurrent.FutureTask;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 饱和策略：当线程池和队列都满了，将剩余任务取消
 * PS ThreadPoolExecutor.DiscardPolicy饱和策略，如果使用Future会无法停止线程，永远在等待结果
 * @author w
 * @date 2020-12-09
 */
public class DiscardPolicyReject implements RejectedExecutionHandler {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        if(!executor.isShutdown()){
            if(r instanceof FutureTask){
                FutureTask task = (FutureTask)r;
                // 会抛出CancellationException
                task.cancel(true);
            }
        }
    }
}

线程工厂接口

package com.panda.service.pool.service;

import java.util.concurrent.ExecutorService;

/**
 * 线程工厂接口
 * @author w
 * @date 2020-12-08
 */
public interface ThreadFactoryService<P> {

    /**
     * 执行任务(自动线程池)
     * @param publicInfo 公共信息
     */
    void execute(P publicInfo) throws Exception;

    /**
     * 执行任务(指定线程池)
     * @param publicInfo 公共信息
     * @param pool 指定线程池
     */
    void execute(P publicInfo, ExecutorService pool) throws Exception;
}

抽象线程工厂

package com.panda.service.pool.service;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

/**
 * 线程工厂
 * @param <P> 公共信息
 * @param <R> 执行任务返回结果
 * @author w
 * @date 2020-12-09
 */
@Slf4j
public abstract class AbstractThreadFactoryService<P,R> implements ThreadFactoryService<P> {

    /**
     * 执行线程任务(自动线程池)
     * @param publicInfo 公共信息
     */
    @Override
    public void execute(P publicInfo) throws Exception{
        execute(publicInfo, null);
    }

    /**
     * 执行线程任务
     * @param publicInfo 公共信息
     * @param pool 指定线程池，若设置null则自动调用buildPool构建线程池并且自动销毁
     */
    @Override
    public void execute(P publicInfo, ExecutorService pool) throws Exception{
        // 线程任务
        List<Callable<R>> threads = null;
        // 任务结果集
        List<Future<R>> futures = null;

        // 是否自动销毁线程池
        boolean poolDestroyFlag = null == pool;

        try{
            // 执行任务之前做点什么
            if(!before(publicInfo)){
                log.info("before method returns false.");
                return;
            }

            // 创建任务
            threads = createTask(publicInfo);
            if(null == threads || threads.size() == 0){
                log.info("No task.");
                return;
            }

            // 构建线程池
            if(poolDestroyFlag){
                pool = buildPool(publicInfo);
                Optional.ofNullable(pool).orElseThrow(() -> new Exception("ExecutorService is empty!"));
            }

            // 执行任务
            futures = new ArrayList<>();
            for(Callable<R> thread : threads){
                // 提交任务
                Future<R> future = pool.submit(thread);
                // 保存任务结果
                futures.add(future);
            }

            // 轮询结果，此处会阻塞主线程，等待所有任务处理完成
            for(Future<R> future : futures){
                R response = null;
                try{
                    // 获取执行结果
                    response = future.get();
                }catch (Exception e){
                    // 执行任务结果处理（异常处理）
                    doResult(publicInfo, e);
                }

                // 执行任务结果处理
                if(null != response){
                    doResult(publicInfo, response);
                }
            }

            // 执行任务之后做点什么
            if(after(publicInfo)){
                log.info("after method returns false.");
            }

        }finally {
            if(null != pool){
                log.info("pool shutdown.");
                pool.shutdown();
            }

            if(null != threads){
                log.info("threads clear.");
                threads.clear();
            }

            if(null != futures){
                log.info("futures clear.");
                futures.clear();
            }
        }
    }

    /**
     * 构建线程池
     * @param publicInfo 公共信息
     * @return 线程池
     */
    protected abstract ExecutorService buildPool(P publicInfo);

    /**
     * 执行任务之前做点什么
     * @param publicInfo 公共信息
     * @return boolean 是否继续执行
     */
    protected abstract boolean before(P publicInfo);

    /**
     * 创建任务
     * @param publicInfo 公共信息
     * @return 任务集
     */
    protected abstract List<Callable<R>> createTask(P publicInfo);

    /**
     * 任务结果处理
     * @param publicInfo 公共信息
     * @param response 处理结果
     */
    protected abstract void doResult(P publicInfo, R response);

    /**
     * 任务结果处理(异常时)
     * @param publicInfo 公共信息
     * @param exception 异常信息
     */
    protected abstract void doResult(P publicInfo, Exception exception);

    /**
     * 执行任务之后做点什么
     * @param publicInfo 公共信息
     * @return boolean 是否继续执行
     */
    protected abstract boolean after(P publicInfo);
}

<调用例子>

新增多线程业务

package com.panda.service.test.task.thread;

import com.alibaba.fastjson.JSONObject;
import com.panda.service.test.task.vo.OrderCheckInfo;
import com.panda.service.test.task.vo.OrderCheckResponse;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Callable;

/**
 * 订单检查异步线程处理
 * @author w
 * @date 2020-12-09
 */
@Slf4j
public class OrderCheckThread implements Callable<OrderCheckResponse> {

    /** 订单信息 **/
    private OrderCheckInfo orderCheckInfo;

    public OrderCheckThread(OrderCheckInfo orderCheckInfo){
        this.orderCheckInfo = orderCheckInfo;
    }

    @Override
    public OrderCheckResponse call() {
        try{
            log.info("开始执行订单检查:{}", JSONObject.toJSONString(orderCheckInfo));
            return new OrderCheckResponse(true, "交易成功", orderCheckInfo.getOrderId());
        }catch (Exception e){
            log.error(e.getMessage(),e);
            return new OrderCheckResponse(false, "交易异常" + e.getMessage(), orderCheckInfo.getOrderId());
        }
    }
}

相关vo类

package com.panda.service.test.task.vo;

import lombok.AllArgsConstructor;
import lombok.Data;

/**
 * 订单检查信息
 */
@AllArgsConstructor
@Data
public class OrderCheckInfo {
    private String orderId;
}


package com.panda.service.test.task.vo;

import lombok.AllArgsConstructor;
import lombok.Data;

/**
 * 订单检查响应结果
 */
@AllArgsConstructor
@Data
public class OrderCheckResponse {
    private boolean status;
    private String result;
    private String orderId;
}

package com.panda.service.test.task.vo;

import lombok.AllArgsConstructor;
import lombok.Data;

/**
 * 会话信息
 */
@AllArgsConstructor
@Data
public class SessionInfo {
    private String oprId;
    private String oprName;
}

实现抽象工厂接口，个性化业务处理

package com.panda.service.test.task.service;

import com.alibaba.fastjson.JSONObject;
import com.panda.service.pool.config.ThreadFactoryConfig;
import com.panda.service.pool.factory.NameThreadFactoryBuilder;
import com.panda.service.pool.factory.ThreadFactoryBuilder;
import com.panda.service.pool.reject.DiscardPolicyReject;
import com.panda.service.pool.service.AbstractThreadFactoryService;
import com.panda.service.test.task.thread.OrderCheckThread;
import com.panda.service.test.task.vo.OrderCheckInfo;
import com.panda.service.test.task.vo.OrderCheckResponse;
import com.panda.service.test.task.vo.SessionInfo;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * 订单检查
 * @author w
 * @date 2020-12-09
 */
@Slf4j
public class OrderCheckService extends AbstractThreadFactoryService<SessionInfo, OrderCheckResponse> {
    @Override
    protected ExecutorService buildPool(SessionInfo sessionInfo) {
        ThreadFactoryConfig config = new ThreadFactoryConfig();
        config.setCorePoolSize(10);
        config.setMaximumPoolSize(20);
        config.setKeepAliveTime(0L);
        config.setTimeUnit(TimeUnit.MILLISECONDS);
        // 采用有界队列
        config.setBlockingQueue(new LinkedBlockingQueue<>(10));
        // 饱和策略采用DiscardPolicy，剩下的任务丢掉不执行
        config.setRejectedExecutionHandler(new DiscardPolicyReject());
        config.setThreadFactory(NameThreadFactoryBuilder.buildThreadFactory("order-check"));
        return ThreadFactoryBuilder.buildThreadPool(config);
    }

    @Override
    protected boolean before(SessionInfo sessionInfo) {
        log.info("在执行任务之前做点什么吧");
        return true;
    }

    @Override
    protected List<Callable<OrderCheckResponse>> createTask(SessionInfo sessionInfo) {
        List<Callable<OrderCheckResponse>> tasks = new ArrayList<>();
        // TODO 测试
        for(int i=0;i<50;i++){
            tasks.add(new OrderCheckThread(new OrderCheckInfo(System.currentTimeMillis() + "_" + i)));
        }
        return tasks;
    }

    @Override
    protected void doResult(SessionInfo sessionInfo, OrderCheckResponse response) {
        log.info("执行结果: {}", JSONObject.toJSONString(response));
    }

    @Override
    protected void doResult(SessionInfo sessionInfo, Exception exception) {
        if(exception instanceof CancellationException){
            log.info("执行异常结果: 线程池和队列都满了，触发饱和策略！");
        }else{
            log.info("执行异常结果: {}",exception.getMessage());
        }
    }

    @Override
    protected boolean after(SessionInfo sessionInfo) {
        log.info("在执行任务之后做点什么吧");
        return true;
    }
}

执行结果

22:46:02.580 [main] INFO com.panda.service.test.task.service.OrderCheckService - 在执行任务之前做点什么吧
22:46:02.722 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_27"}
22:46:02.723 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_0"}
22:46:02.724 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_7"}
22:46:02.724 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_24"}
22:46:02.723 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_20"}
22:46:02.723 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_25"}
22:46:02.723 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_22"}
22:46:02.723 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_26"}
22:46:02.723 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_9"}
22:46:02.723 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_4"}
22:46:02.723 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_3"}
22:46:02.723 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_28"}
22:46:02.722 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_1"}
22:46:02.723 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_8"}
22:46:02.725 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_12"}
22:46:02.725 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_13"}
22:46:02.725 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_15"}
22:46:02.725 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_17"}
22:46:02.725 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_19"}
22:46:02.724 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_21"}
22:46:02.725 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_18"}
22:46:02.724 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_23"}
22:46:02.723 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_2"}
22:46:02.722 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_29"}
22:46:02.725 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_11"}
22:46:02.724 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_10"}
22:46:02.725 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_14"}
22:46:02.724 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_5"}
22:46:02.725 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_16"}
22:46:02.723 [order-check] INFO com.panda.service.test.task.thread.OrderCheckThread - 开始执行订单检查:{"orderId":"1607784362586_6"}
22:46:02.730 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_0","result":"交易成功","status":true}
22:46:02.730 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_1","result":"交易成功","status":true}
22:46:02.730 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_2","result":"交易成功","status":true}
22:46:02.730 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_3","result":"交易成功","status":true}
22:46:02.730 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_4","result":"交易成功","status":true}
22:46:02.730 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_5","result":"交易成功","status":true}
22:46:02.730 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_6","result":"交易成功","status":true}
22:46:02.730 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_7","result":"交易成功","status":true}
22:46:02.730 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_8","result":"交易成功","status":true}
22:46:02.730 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_9","result":"交易成功","status":true}
22:46:02.730 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_10","result":"交易成功","status":true}
22:46:02.730 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_11","result":"交易成功","status":true}
22:46:02.730 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_12","result":"交易成功","status":true}
22:46:02.730 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_13","result":"交易成功","status":true}
22:46:02.730 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_14","result":"交易成功","status":true}
22:46:02.730 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_15","result":"交易成功","status":true}
22:46:02.730 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_16","result":"交易成功","status":true}
22:46:02.730 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_17","result":"交易成功","status":true}
22:46:02.731 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_18","result":"交易成功","status":true}
22:46:02.731 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_19","result":"交易成功","status":true}
22:46:02.731 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_20","result":"交易成功","status":true}
22:46:02.731 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_21","result":"交易成功","status":true}
22:46:02.731 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_22","result":"交易成功","status":true}
22:46:02.731 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_23","result":"交易成功","status":true}
22:46:02.731 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_24","result":"交易成功","status":true}
22:46:02.731 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_25","result":"交易成功","status":true}
22:46:02.731 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_26","result":"交易成功","status":true}
22:46:02.731 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_27","result":"交易成功","status":true}
22:46:02.731 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_28","result":"交易成功","status":true}
22:46:02.731 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行结果: {"orderId":"1607784362586_29","result":"交易成功","status":true}
22:46:02.731 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.731 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.731 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.731 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.731 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.732 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.732 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.734 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.734 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.734 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.734 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.735 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.735 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.735 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.735 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.735 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.735 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.735 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.735 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.735 [main] INFO com.panda.service.test.task.service.OrderCheckService - 执行异常结果: 线程池和队列都满了，触发饱和策略！
22:46:02.735 [main] INFO com.panda.service.test.task.service.OrderCheckService - 在执行任务之后做点什么吧
22:46:02.735 [main] INFO com.panda.service.pool.service.AbstractThreadFactoryService - after method returns false.
22:46:02.735 [main] INFO com.panda.service.pool.service.AbstractThreadFactoryService - pool shutdown.
22:46:02.735 [main] INFO com.panda.service.pool.service.AbstractThreadFactoryService - threads clear.
22:46:02.735 [main] INFO com.panda.service.pool.service.AbstractThreadFactoryService - futures clear.

Process finished with exit code 0

除特别声明，本站所有文章均为原创，如需转载请以超级链接形式注明出处：SmartCat's Blog

标签：Java ThreadPoolExecutor 线程池 线程池 多线程 并发编程 java 队列

上一篇： redis 基础知识

下一篇： 练习23(名称空间，作用域，取值顺序)

精华推荐