spring-boot @Async 的使用、自定义Execut

spring-boot @Async 的使用、自定义Executor的配置方法

异步使用场景:发送短信,发送邮件,App消息推送,运维发布任务

基本类

1. 1. TaskExecutor
      Spring异步线程池的接口类，其实质是java.util.concurrent.Executor

Spring 已经实现的异常线程池：

1. SimpleAsyncTaskExecutor：不是真的线程池，这个类不重用线程，每次调用都会创建一个新的线程。

2. SyncTaskExecutor：这个类没有实现异步调用，只是一个同步操作。只适用于不需要多线程的地方

3. ConcurrentTaskExecutor：Executor的适配类，不推荐使用。如果ThreadPoolTaskExecutor不满足要求时，才用考虑使用这个类

4. SimpleThreadPoolTaskExecutor：是Quartz的SimpleThreadPool的类。线程池同时被quartz和非quartz使用，才需要使用此类

5. ThreadPoolTaskExecutor ：最常使用，推荐。 其实质是对java.util.concurrent.ThreadPoolExecutor的包装

6. @Async
   spring对过@Async定义异步任务

异步的方法有3种

1. 最简单的异步调用，返回值为void
2. 带参数的异步调用 异步方法可以传入参数
3. 异常调用返回Future

实现

1. ExecutorPoolConfig(线程池配置类)

/**
 * 异步使用的线程池
 */
@EnableAsync// 启动异步调用
@Configuration
public class ExecutorPoolConfig {
    @Bean("executor")
    public Executor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(20);
        executor.setMaxPoolSize(50);
        executor.setQueueCapacity(5000);
        executor.setKeepAliveSeconds(120);
        executor.setThreadNamePrefix("task-");
        /**
         * AbortPolicy（默认）：丢弃任务，抛出RejectedExecutionException异常
         * DiscardPolicy：也是丢弃任务，但不抛出异常
         * DiscardOldestPolicy：对被拒绝的任务（新来的）不抛弃，而是抛弃队列里面等待最久的一个任务，然后把该拒绝任务加到队列
         * CallerRunsPolicy：重试添加当前被拒绝的任务，他会自动重复调用 execute() 方法，直到成功。
         */
        //executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
        return executor;
    }
}

2. 异步实现业务（推送es）

/**
 * 异步任务入口
 *
 * @author wuby
 * @date 2021-08-30
 */
@Lazy
@Slf4j
@Component(value = "taskExecutor")
public class TaskExecutor {
    @Resource
    private MqService mqService;

    /**
     * 券状态变更推送ES
     */
    @Async("executor")
    public void asyncSendCouponStatusToEs(String couponNum, String uid, Integer opt, Integer status, String orderId, Integer isSuper, Integer superCpIndex) {
        try {
            mqService.sendCouponStatusChangeToMq(couponNum, uid, opt, status, orderId, isSuper, superCpIndex);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }
}

3. 调用的地方

/**
 * <p>
 * 有赞推送mq
 * </p>
 */
@RestController
@RequestMapping("/mqApi")
@Slf4j
public class MqApi {

    @Autowired
    private TaskExecutor taskExecutor;

    /**
     * @param couponBO
     * @return
     */
    @RequestMapping("/asyncSendCouponStatus")
    public RespDTO asyncSendCouponStatus(@RequestBody ReqDTO<ReceivedCouponBO> couponBO) {
        ReceivedCouponBO reqDTO = couponBO.parseBizParams(ReceivedCouponBO.class); //注解上面进行了参数判断
        log.info("有赞传送的数据为--{}", JacksonUtils.obj2json(reqDTO));
        if (reqDTO == null) {
            return RespDTO.getFail(Constants.RESPONSE_CODE_240001, "券状态变更推送ES失败", Constants.RESPONSE_CODE_240001, "券状态变更推送ES失败");
        }
        if (StringUtils.isEmpty(reqDTO.getCouponAtno())) {
            return RespDTO.getFail(Constants.RESPONSE_CODE_240001, "券编号不能为空", Constants.RESPONSE_CODE_240001, "券编号不能为空");
        }
        if (StringUtils.isEmpty(reqDTO.getUid())) {
            return RespDTO.getFail(Constants.RESPONSE_CODE_240001, "用户id不能为空", Constants.RESPONSE_CODE_240001, "用户id不能为空");
        }
        if (reqDTO.getOpt() == null) {
            return RespDTO.getFail(Constants.RESPONSE_CODE_240001, "操作状态不能为空", Constants.RESPONSE_CODE_240001, "操作状态不能为空");
        }
        if (reqDTO.getStatus() == null) {
            return RespDTO.getFail(Constants.RESPONSE_CODE_240001, "券状态不能为空", Constants.RESPONSE_CODE_240001, "券状态不能为空");
        }
        taskExecutor.asyncSendCouponStatusToEs(reqDTO.getCouponAtno(), reqDTO.getUid(), reqDTO.getOpt(), reqDTO.getStatus(),
                reqDTO.getOrderId(), reqDTO.getIsSuper(), reqDTO.getSuperCpIndex());
        return RespDTO.getSucc();
    }

}

参数说明

线程池参数说明

1. ‘id’ ： 线程的名称的前缀
2. ‘pool-size’：线程池的大小。支持范围”min-max”和固定值（此时线程池core和max sizes相同）
3. ‘queue-capacity’ ：排队队列长度
   ○ The main idea is that when a task is submitted, the executor will first try to use a free thread if the number of active threads is currently less than the core size.
   ○ If the core size has been reached, then the task will be added to the queue as long as its capacity has not yet been reached.
   ○ Only then, if the queue’s capacity has been reached, will the executor create a new thread beyond the core size.
   ○ If the max size has also been reached, then the executor will reject the task.
   ○ By default, the queue is unbounded, but this is rarely the desired configuration because it can lead to OutOfMemoryErrors if enough tasks are added to that queue while all pool threads are busy.
4. ‘rejection-policy’: 对拒绝的任务处理策略
   ○ In the default ThreadPoolExecutor.AbortPolicy, the handler throws a runtime RejectedExecutionException upon rejection.
   ○ In ThreadPoolExecutor.CallerRunsPolicy, the thread that invokes execute itself runs the task. This provides a simple feedback control mechanism that will slow down the rate that new tasks are submitted.
   ○ In ThreadPoolExecutor.DiscardPolicy, a task that cannot be executed is simply dropped.
   ○ In ThreadPoolExecutor.DiscardOldestPolicy, if the executor is not shut down, the task at the head of the work queue is dropped, and then execution is retried (which can fail again, causing this to be repeated.)
5. ‘keep-alive’ ： 线程保活时间（单位秒）
   setting determines the time limit (in seconds) for which threads may remain idle before being terminated. If there are more than the core number of threads currently in the pool, after waiting this amount of time without processing a task, excess threads will get terminated. A time value of zero will cause excess threads to terminate immediately after executing a task without remaining follow-up work in the task queue()