当前位置：首页 > article >正文

【最佳实践】配置类封装-Async异步注解以及自定义线程池

article 2024/9/22 5:44:49

效果是：能点进去看到自定义的线程池，代表指定自定义的线程池成功！

自定义Async线程池

自定义线程池

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.ThreadPoolExecutor;

/**
 * 线程池配置
 */
@Configuration
@Slf4j
public class ThreadPoolConfig {
    @Value("${asyncThreadPool.corePoolSize}")
    private int corePoolSize;

    @Value("${asyncThreadPool.maxPoolSize}")
    private int maxPoolSize;

    @Value("${asyncThreadPool.queueCapacity}")
    private int queueCapacity;

    @Value("${asyncThreadPool.keepAliveSeconds}")
    private int keepAliveSeconds;

    @Value("${asyncThreadPool.awaitTerminationSeconds}")
    private int awaitTerminationSeconds;

    @Value("${asyncThreadPool.threadNamePrefix}")
    private String threadNamePrefix;

    /**
     * 线程池配置
     * @return
     */
    @Bean(name = "threadPoolTaskExecutor")
    public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
        log.info("---------- 线程池开始加载 ----------");
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        threadPoolTaskExecutor.setCorePoolSize(corePoolSize);                        // 核心线程池大小
        threadPoolTaskExecutor.setMaxPoolSize(maxPoolSize);                          // 最大线程数
        threadPoolTaskExecutor.setQueueCapacity(keepAliveSeconds);                        // 队列容量
        threadPoolTaskExecutor.setKeepAliveSeconds(queueCapacity);                        // 活跃时间
        //threadPoolTaskExecutor.setAwaitTerminationSeconds(awaitTerminationSeconds);        // 主线程等待子线程执行时间
        threadPoolTaskExecutor.setThreadNamePrefix(threadNamePrefix);        // 线程名字前缀
        // RejectedExecutionHandler:当pool已经达到max-size的时候，如何处理新任务
        //      CallerRunsPolicy:不在新线程中执行任务，而是由调用者所在的线程来执行
        threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        threadPoolTaskExecutor.initialize(); // 初始化
        log.info("---------- 线程池加载完成 ----------");
        return threadPoolTaskExecutor;
    }
}

application.yml 配置

# @async 线程池配置
asyncThreadPool:
  corePoolSize: 10
  maxPoolSize: 50
  queueCapacity: 1000
  keepAliveSeconds: 60
  awaitTerminationSeconds: 600
  threadNamePrefix: asyncTask-

标准模板：优雅的写法

将配置文件（默认配置）与业务代码分离

MyAsyncThreadPoolExecutorProperties

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;

@Data
@ConfigurationProperties(prefix = "async.thread.pool")
public class ThreadPoolTaskExecutorProperties {
    /**
     * 核心线程数
     */
    private int corePoolSize = 10;

    /**
     * 最大线程数
     */
    private int maxPoolSize = 50;

    /**
     * 队列容量
     */
    private int queueCapacity = 1000;

    /**
     * 线程存活时间
     */
    private int keepAliveSeconds = 60;

    /**
     * 等待终止时间
     */
    private int awaitTerminationSeconds = 600;

    /**
     * 线程名称前缀
     */
    private String threadNamePrefix = "asyncTask-";
}

MyAsyncThreadPoolExecutor

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@Configuration
@EnableAsync
@EnableConfigurationProperties(ThreadPoolTaskExecutorProperties.class)
public class MyAsyncThreadPoolExecutor {

    private final ThreadPoolTaskExecutorProperties properties;

    @Autowired
    public MyAsyncThreadPoolExecutor(ThreadPoolTaskExecutorProperties properties) {
        this.properties = properties;
    }
    @Bean("threadPoolTaskExecutor")
    public ThreadPoolTaskExecutor threadPoolExecutor() {
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        threadPoolTaskExecutor.setCorePoolSize(properties.getCorePoolSize());
        threadPoolTaskExecutor.setMaxPoolSize(properties.getMaxPoolSize());
        threadPoolTaskExecutor.setQueueCapacity(properties.getQueueCapacity());
        threadPoolTaskExecutor.setKeepAliveSeconds(properties.getKeepAliveSeconds());
        threadPoolTaskExecutor.setAwaitTerminationSeconds(properties.getAwaitTerminationSeconds());
        threadPoolTaskExecutor.setThreadNamePrefix(properties.getThreadNamePrefix());
        threadPoolTaskExecutor.initialize();
        return threadPoolTaskExecutor;
    }
}

测试使用

使用方法：可以指定或者不指定，最好指定！

    /**
     * @Async失效情况： TODO 异步发送方案OOM问题
     */
    @Async("threadPoolTaskExecutor")
    public void testSend(String site) {
        ///notify/v1/send-code?name=http://47.98.233.38:5212/login
        if (StrUtil.isBlank(site)) {
            site = "https://www.xdclass.net/";
        }

        // 发送验证码模拟
        ResponseEntity<String> forEntity = this.restTemplate.getForEntity(site, String.class);
        String body = forEntity.getBody();
        log.info(body);
    }

特征是：会显示用了哪个线程完成任务，但是不会呈现内部的worker执行细节（得看源码）

注解失效问题

【官方】：如下方式会使@Async失效

一、异步方法使用static修饰
二、异步类没有使用@Component注解（或其他注解）导致spring无法扫描到异步类
三、异步方法不能与被调用的异步方法在同一个类中
四、类中需要使用@Autowired或@Resource等注解自动注入，不能自己手动new对象
五、如果使用SpringBoot框架必须在启动类中增加@EnableAsync注解

成功使用的原则是：

因为底层是由动态代理实现的，所以需要被spring管理
由spring管理，所以里面的类引用都是**@Autowired或@Resource等注解⾃动注⼊**，全程都要在spring的管辖范围内（方便使用动态代理的方式来完成异步调用），即调用者也必须是spring管理范围内的组件，不允许手动new对象
方法要求：public、非static非共享、返回值为void、Futrue

特点：

底层还有一个线程池，要么自己实现，要么spring自带（超不推荐）
动态代理实现

其他

简单demo写法

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@Configuration
@EnableAsync
public class MyAsyncThreadPoolExecutor {
    @Bean("threadPoolTaskExecutor")
    public ThreadPoolTaskExecutor threadPoolExecutor() {
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        threadPoolTaskExecutor.setCorePoolSize(16);
        threadPoolTaskExecutor.setMaxPoolSize(64);
        threadPoolTaskExecutor.setQueueCapacity(1024);
        threadPoolTaskExecutor.setKeepAliveSeconds(60);
        threadPoolTaskExecutor.setAwaitTerminationSeconds(600);
        threadPoolTaskExecutor.setThreadNamePrefix("自定义线程池-");
        threadPoolTaskExecutor.initialize();
        return threadPoolTaskExecutor;
    }
}