springboot原生线程池使用、异步处理

谁借莪１个温暖的怀抱￠ 2022-04-24 00:56 198阅读 0赞

1、启动类添加注解，开启异步支持

@EnableAsync//开启异步处理支持，开启springboot线程池

2、配置文件配置属性（我是自定义的配置文件，config.properties，也可以配置在application.properties中）

#线程池配置
    #核心线程池大小
    async.pool.corePoolSize=20
    #最大线程数
    async.pool.maxPoolSize=40
    #活跃时间
    async.pool.keepAliveSeconds=300
    #队列容量
    async.pool.queueCapacity=50

3、实现原生异步配置类，并配置自定义参数

AsyncExecuteThreadPool

其中：@Slf4j是用来打印日志的。@PropertySource(value="classpath:config.properties")是用来读取自定义配置文件的，如果配置信息放在了application.properties中无需使用。

package com.gili.CPMasterController.common.async;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.annotation.PropertySource;
    import org.springframework.scheduling.annotation.AsyncConfigurer;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
    import java.lang.reflect.Method;
    import java.util.concurrent.Executor;
    import java.util.concurrent.ThreadPoolExecutor;
    
    /**
     * 原生(Spring)异步任务线程池装配类
     * Created by groot
     */
    @Slf4j
    @Configuration
    @PropertySource(value="classpath:config.properties")
    public class AsyncExecuteThreadPool implements AsyncConfigurer {
    
        @Value("${async.pool.corePoolSize}")
        private int corePoolSize;//核心线程池大小
    
        @Value("${async.pool.maxPoolSize}")
        private int maxPoolSize;//最大线程数
    
        @Value("${async.pool.keepAliveSeconds}")
        private int keepAliveSeconds;//活跃时间
    
        @Value("${async.pool.queueCapacity}")
        private int queueCapacity;//队列容量
    
        @Override
        public Executor getAsyncExecutor() {
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            //核心线程池大小
            executor.setCorePoolSize(corePoolSize);
            //最大线程数
            executor.setMaxPoolSize(maxPoolSize);
            //队列容量
            executor.setQueueCapacity(queueCapacity);
            //活跃时间
            executor.setKeepAliveSeconds(keepAliveSeconds);
            //线程名字前缀
            executor.setThreadNamePrefix("CP Thread-");
    
            // setRejectedExecutionHandler：当pool已经达到max size的时候，如何处理新任务
            // CallerRunsPolicy：不在新线程中执行任务，而是由调用者所在的线程来执行
            executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            executor.initialize();
            return executor;
        }
    
    
        /**
         *  异步任务中异常处理
         * @return
         */
        @Override
        public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
            return new AsyncUncaughtExceptionHandler() {
    
                @Override
                public void handleUncaughtException(Throwable arg0, Method arg1, Object... arg2) {
                    log.error("=========================="+arg0.getMessage()+"=======================", arg0);
                    log.error("exception method:"+arg1.getName());
                }
            };
        }
    }

4、线程池的使用，编写异步任务业务类（注意：异步方法不能与被调用的异步方法在同一个类中，否则异步无效）

AsyncTaskServiceDemo

package com.gili.CPMasterController.common.async;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.scheduling.annotation.Async;
    import org.springframework.stereotype.Service;
    
    /**
     * @Author : groot
     * 2019年4月11日 13:28:58
     *  异步执行线程池获取线程demo
     *  注意：异步方法不能与被调用的异步方法在同一个类中，否则异步无效
     *  使用方法：1、定义实现业务逻辑的方法如：doTask(int i)
     *            2、在方法上架注解 @Async，这样就会在执行方法时从原生线程池中获取线程来异步执行此方法
     */
    @Service
    @Slf4j
    public class AsyncTaskServiceDemo {
    
        @Async
        public void doTask(int i) throws InterruptedException{
            log.info("Task-"+i+" started.");
            if(i%3==0){
                log.info("i "+i+" sleep start...");
                Thread.sleep(200);
                log.info("i "+i+" sleep end...");
            }
        }
    }

5、测试

在测试类中添加测试方法

package com.gili.CPMasterController;
    
    import com.gili.CPMasterController.common.async.AsyncTaskServiceDemo;
    import lombok.extern.slf4j.Slf4j;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    import java.util.concurrent.ExecutionException;
    
    @RunWith(SpringRunner.class)
    @SpringBootTest
    @Slf4j
    public class CPApplicationTests {
    
        @Autowired
        AsyncTaskServiceDemo asyncTask;
    
        /**
         * 线程异步调用处理测试
         * @throws InterruptedException
         * @throws ExecutionException
         */
        @Test
        public void AsyncTaskNativeTest() throws InterruptedException, ExecutionException {
    
            for (int i = 0; i < 20; i++) {
                asyncTask.doTask(i);
            }
            log.info("All tasks finished.");
        }
    
    }

6、执行测试方法AsyncTaskNativeTest()得到以下日志：

14:00:37.434 [main] DEBUG o.s.t.c.s.AbstractDirtiesContextTestExecutionListener - Before test method: context [DefaultTestContext@398dada8 testClass = CPApplicationTests, testInstance = com.gili.CPMasterController.CPApplicationTests@17d816b3, testMethod = AsyncTaskNativeTest@CPApplicationTests, testException = [null], mergedContextConfiguration = [WebMergedContextConfiguration@7cb502c testClass = CPApplicationTests, locations = '{}', classes = '{class com.gili.CPMasterController.Application}', contextInitializerClasses = '[]', activeProfiles = '{}', propertySourceLocations = '{}', propertySourceProperties = '{org.springframework.boot.test.context.SpringBootTestContextBootstrapper=true}', contextCustomizers = set[org.springframework.boot.test.autoconfigure.properties.PropertyMappingContextCustomizer@0, org.springframework.boot.test.autoconfigure.web.servlet.WebDriverContextCustomizerFactory$Customizer@4988d8b8, org.springframework.boot.test.context.filter.ExcludeFilterContextCustomizer@609cd4d8, org.springframework.boot.test.json.DuplicateJsonObjectContextCustomizerFactory$DuplicateJsonObjectContextCustomizer@13b6aecc, org.springframework.boot.test.mock.mockito.MockitoContextCustomizer@0, org.springframework.boot.test.web.client.TestRestTemplateContextCustomizer@49e5f737], resourceBasePath = 'src/main/webapp', contextLoader = 'org.springframework.boot.test.context.SpringBootContextLoader', parent = [null]], attributes = map['org.springframework.test.context.web.ServletTestExecutionListener.activateListener' -> true, 'org.springframework.test.context.web.ServletTestExecutionListener.populatedRequestContextHolder' -> true, 'org.springframework.test.context.web.ServletTestExecutionListener.resetRequestContextHolder' -> true]], class annotated with @DirtiesContext [false] with mode [null], method annotated with @DirtiesContext [false] with mode [null].
    14:00:37.445 [main] DEBUG o.s.t.c.c.DefaultCacheAwareContextLoaderDelegate - Retrieved ApplicationContext from cache with key [[WebMergedContextConfiguration@7cb502c testClass = CPApplicationTests, locations = '{}', classes = '{class com.gili.CPMasterController.Application}', contextInitializerClasses = '[]', activeProfiles = '{}', propertySourceLocations = '{}', propertySourceProperties = '{org.springframework.boot.test.context.SpringBootTestContextBootstrapper=true}', contextCustomizers = set[org.springframework.boot.test.autoconfigure.properties.PropertyMappingContextCustomizer@0, org.springframework.boot.test.autoconfigure.web.servlet.WebDriverContextCustomizerFactory$Customizer@4988d8b8, org.springframework.boot.test.context.filter.ExcludeFilterContextCustomizer@609cd4d8, org.springframework.boot.test.json.DuplicateJsonObjectContextCustomizerFactory$DuplicateJsonObjectContextCustomizer@13b6aecc, org.springframework.boot.test.mock.mockito.MockitoContextCustomizer@0, org.springframework.boot.test.web.client.TestRestTemplateContextCustomizer@49e5f737], resourceBasePath = 'src/main/webapp', contextLoader = 'org.springframework.boot.test.context.SpringBootContextLoader', parent = [null]]]
    14:00:37.445 [main] DEBUG o.springframework.test.context.cache - Spring test ApplicationContext cache statistics: [DefaultContextCache@40a8a26f size = 1, maxSize = 32, parentContextCount = 0, hitCount = 2, missCount = 1]
    14:00:38.392 [main] INFO  o.s.s.c.ThreadPoolTaskExecutor - Initializing ExecutorService
    14:00:38.430 [CP Thread-11] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-10 started.
    14:00:38.430 [CP Thread-10] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-9 started.
    14:00:38.422 [CP Thread-1] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-0 started.
    14:00:38.430 [CP Thread-9] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-8 started.
    14:00:38.430 [CP Thread-8] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-7 started.
    14:00:38.430 [CP Thread-7] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-6 started.
    14:00:38.431 [CP Thread-10] INFO  c.g.C.c.async.AsyncTaskServiceDemo - i 9 sleep start...
    14:00:38.430 [CP Thread-5] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-4 started.
    14:00:38.429 [CP Thread-4] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-3 started.
    14:00:38.435 [CP Thread-4] INFO  c.g.C.c.async.AsyncTaskServiceDemo - i 3 sleep start...
    14:00:38.429 [CP Thread-3] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-2 started.
    14:00:38.429 [CP Thread-2] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-1 started.
    14:00:38.429 [CP Thread-20] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-19 started.
    14:00:38.424 [CP Thread-19] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-18 started.
    14:00:38.424 [CP Thread-18] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-17 started.
    14:00:38.437 [CP Thread-19] INFO  c.g.C.c.async.AsyncTaskServiceDemo - i 18 sleep start...
    14:00:38.423 [CP Thread-13] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-12 started.
    14:00:38.438 [CP Thread-13] INFO  c.g.C.c.async.AsyncTaskServiceDemo - i 12 sleep start...
    14:00:38.424 [CP Thread-17] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-16 started.
    14:00:38.424 [CP Thread-16] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-15 started.
    14:00:38.439 [CP Thread-16] INFO  c.g.C.c.async.AsyncTaskServiceDemo - i 15 sleep start...
    14:00:38.424 [CP Thread-15] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-14 started.
    14:00:38.423 [CP Thread-14] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-13 started.
    14:00:38.434 [CP Thread-1] INFO  c.g.C.c.async.AsyncTaskServiceDemo - i 0 sleep start...
    14:00:38.434 [CP Thread-7] INFO  c.g.C.c.async.AsyncTaskServiceDemo - i 6 sleep start...
    14:00:38.423 [CP Thread-12] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-11 started.
    14:00:38.430 [CP Thread-6] INFO  c.g.C.c.async.AsyncTaskServiceDemo - Task-5 started.
    14:00:39.033 [CP Thread-16] INFO  c.g.C.c.async.AsyncTaskServiceDemo - i 15 sleep end...
    14:00:39.033 [CP Thread-13] INFO  c.g.C.c.async.AsyncTaskServiceDemo - i 12 sleep end...
    14:00:39.033 [CP Thread-19] INFO  c.g.C.c.async.AsyncTaskServiceDemo - i 18 sleep end...
    14:00:39.034 [CP Thread-7] INFO  c.g.C.c.async.AsyncTaskServiceDemo - i 6 sleep end...
    14:00:39.034 [CP Thread-4] INFO  c.g.C.c.async.AsyncTaskServiceDemo - i 3 sleep end...
    14:00:39.034 [CP Thread-10] INFO  c.g.C.c.async.AsyncTaskServiceDemo - i 9 sleep end...
    14:00:39.034 [CP Thread-1] INFO  c.g.C.c.async.AsyncTaskServiceDemo - i 0 sleep end...

通过日志可以明显看出每一次调用异步处理方法doTask时都是获取了不同的线程，如CP Thread-19。并且是异步执行的。

demo如此简单，具体业务逻辑可以在这个基础之上起飞了。祝，同行们，Happy everyday！