Java异步注解@Async结合自定义线程池的使用

妖狐艹你老母 2023-10-04 08:10 3阅读 0赞

#### 一、背景 ####

项目中已经存在若干个线程池，为了保证业务之间互不影响，不会因为某一个业务的线程使用率极高，导致其余业务无线程可用或处理时间过长，现需采用线程池隔离的思想。

#### 二、具体代码实现 ####

##### 1. 自定义线程池配置类 #####

import lombok.extern.slf4j.Slf4j;
    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.*;
    import java.util.concurrent.atomic.AtomicInteger;
    
    /**
     * 线程池配置类
     * @author saint
     */
    @Configuration
    @EnableAsync
    @Slf4j
    public class ThreadPoolConfig {
        
    
        @Bean("multiGuaranteeExecutor")
        public Executor multiGuaranteeExecutor() {
        
            ThreadPoolTaskExecutor multiGuaranteeExecutor = new ThreadPoolTaskExecutor();
            multiGuaranteeExecutor.setCorePoolSize(10);
            multiGuaranteeExecutor.setMaxPoolSize(20);
            multiGuaranteeExecutor.setQueueCapacity(200);
            multiGuaranteeExecutor.setKeepAliveSeconds(30);
            // 设置线程名称前缀，当程序出现性能问题时，使用jstack命令、JvisualVM、JProfile工具根据线程名称可以快速定位代码问题。
            multiGuaranteeExecutor.setThreadNamePrefix("multi-guarantee-thread--");
            // 使用自定义拒绝策略，保证重要数据不丢失
            multiGuaranteeExecutor.setRejectedExecutionHandler(new CustomRejectedExecutionHandler());
            return multiGuaranteeExecutor;
        }
    
        @Bean("simpleExecutor")
        public Executor simpleExecutor() {
        
            ThreadPoolTaskExecutor simpleExecutor = new ThreadPoolTaskExecutor();
            simpleExecutor.setCorePoolSize(10);
            simpleExecutor.setMaxPoolSize(20);
            simpleExecutor.setQueueCapacity(200);
            simpleExecutor.setKeepAliveSeconds(30);
            simpleExecutor.setThreadNamePrefix("simple-thread--");
            simpleExecutor.setRejectedExecutionHandler(new CustomRejectedExecutionHandler());
            return simpleExecutor;
        }
    
        /**
         * Customize thread factory
         */
        private class CustomThreadFactory implements ThreadFactory {
        
            private AtomicInteger count = new AtomicInteger(1);
    
            @Override
            public Thread newThread(Runnable r) {
        
                Thread t = new Thread(r);
                String threadName = "multi-guarantee-thread-" + count.getAndIncrement();
                t.setName(threadName);
                return t;
            }
        }
    
    
        /**
         * Customize rejected handler
         */
        private class CustomRejectedExecutionHandler implements RejectedExecutionHandler {
        
    
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        
                try {
        
                    // 一直尝试往队列中添加任务
                    // 此处通过打印日志、发MQ等操作保证数据不丢失；减少线程池的工作压力。
                    executor.getQueue().put(r);
                    log.info("multi-guarantee thread executed failure");
                } catch (InterruptedException e) {
        
                    log.error("Exception in putting thread queue again, ", e);
                }
    
            }
        }
    }

##### 2. 自定义线程池的使用 #####

import lombok.extern.slf4j.Slf4j;
    import org.springframework.scheduling.annotation.Async;
    import org.springframework.stereotype.Service;
    
    import java.util.concurrent.TimeUnit;
    
    /**
     * @author xzhou.Saint
     */
    @Slf4j
    @Service
    public class AsyncService {
        
    
        @Async("multiGuaranteeExecutor")
        public void one() throws Exception {
        
            log.info("one");
            TimeUnit.SECONDS.sleep(2);
            log.info("one success");
        }
    
        @Async("multiGuaranteeExecutor")
        public void two() throws Exception {
        
            log.info("two");
            TimeUnit.SECONDS.sleep(2);
            log.info("two success");
        }
    
        @Async("simpleExecutor")
        public void three() throws Exception {
        
            log.info("three");
            TimeUnit.SECONDS.sleep(2);
            log.info("three success");
        }
    }

##### 3. 测试类 #####

@SpringBootTest
    class ApplicationTests {
        
    
        @Autowired
        private AsyncService asyncService;
    
        @Test
        void contextLoads() throws Exception {
        
            asyncService.one();
            asyncService.two();
            asyncService.three();
    
            Thread.currentThread().join();
        }
    
    }

观察日志我们可以发现，两个线程池已经生效。