建议先看我另外一篇文章:并发编程使用了 线程池 ThreadPoolExecutor 程序性能有了质的突破
开发环境
- SpringBoot 2.1.10.RELEASE
- JDK 1.8
1:启动类添加@EnableAsync注解
package com.nobody;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableAsync;
@SpringBootApplication
@EnableAsync
public class Application {
  public static void main(String[] args) {
    SpringApplication.run(Application.class, args);
  }
}2:线程池配置
package com.nobody.config;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
@Configuration
public class ExecutorConfig {
    public static final int CORE_POOL_SIZE = 5;
    public static final int MAX_POOL_SIZE = 10;
    public static final int QUEUE_CAPACITY = 100;
    @Bean("myExecutor")
    public Executor asyncServiceExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 核心线程数大小
        executor.setCorePoolSize(CORE_POOL_SIZE);
        // 最大线程数大小
        executor.setMaxPoolSize(MAX_POOL_SIZE);
        // 阻塞队列容量
        executor.setQueueCapacity(QUEUE_CAPACITY);
        // 线程名前缀
        executor.setThreadNamePrefix("myTask-");
        // rejectionPolicy:当queue达到maxSize并且此时maxPoolSize也达到最大值的时候,对于新任务的处理策略
        // CallerRunsPolicy:不在新线程中执行任务,而是交由调用者所在的线程来执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
}3:编写异步方法
package com.nobody.domain;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Service;
@Service
public class AsyncService {
    private static final Logger LOGGER = LoggerFactory.getLogger(AsyncService.class);
    // 简单版本
    @Async
    public void asyncTaskA() {
        LOGGER.info("asyncTaskA 异步方法执行.");
    }
    // 入参类型,此处将调用者(线程)的MDC信息传入,主要是保持日志输出时,两个线程有同样的traceId
    @Async
    public void asyncTaslB(Map<String, String> contextMap) {
        MDC.setContextMap(contextMap);
        LOGGER.info("asyncTaskB 异步方法执行.");
    }
    // 线程执行后计数器减1
    @Async
    public void asyncTaskC(CountDownLatch cdl) {
        LOGGER.info("asyncTaskC 异步方法执行.");
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        cdl.countDown();
    }
    // 有返回参数类型
    @Async
    public Future<String> asyncTaskD(String message) {
        LOGGER.info("asyncTaskD 异步方法执行.");
        return new AsyncResult<String>("@" + message);
    }
}4:异步调用
package com.nobody.controller;
import java.util.concurrent.ExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import com.nobody.domain.AsyncService;
@RestController
@RequestMapping("demo")
public class DemoController {
    private static final Logger LOGGER = LoggerFactory.getLogger(DemoController.class);
    @Autowired
    private AsyncService asyncService;
    @GetMapping("test")
    public void test() throws InterruptedException, ExecutionException {
        asyncService.asyncTaskA();
        LOGGER.info("主线程执行...");
    }
}输出结果:

package com.nobody.controller;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import com.nobody.domain.AsyncService;
@RestController
@RequestMapping("demo")
public class DemoController {
    private static final Logger LOGGER = LoggerFactory.getLogger(DemoController.class);
    @Autowired
    private AsyncService asyncService;
    @GetMapping("test")
    public void test() throws InterruptedException, ExecutionException {
        MDC.put("traceId", UUID.randomUUID().toString().replaceFirst("-", ""));
        asyncService.asyncTaskB(MDC.getCopyOfContextMap());
        LOGGER.info("主线程执行...");
    }
}输出结果:

package com.nobody.controller;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import com.nobody.domain.AsyncService;
@RestController
@RequestMapping("demo")
public class DemoController {
    private static final Logger LOGGER = LoggerFactory.getLogger(DemoController.class);
    @Autowired
    private AsyncService asyncService;
    @GetMapping("test")
    public void test() throws InterruptedException, ExecutionException {
        CountDownLatch cdl = new CountDownLatch(2);
        asyncService.asyncTaskC(cdl);
        asyncService.asyncTaskC(cdl);
        try {
            // 等等两个线程执行后,再继续下面的执行,如果超过5s则不等待
            cdl.await(5, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        LOGGER.info("主线程执行...");
    }
}输出结果:

package com.nobody.controller;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import com.nobody.domain.AsyncService;
@RestController
@RequestMapping("demo")
public class DemoController {
    private static final Logger LOGGER = LoggerFactory.getLogger(DemoController.class);
    @Autowired
    private AsyncService asyncService;
    @GetMapping("test")
    public void test() throws InterruptedException, ExecutionException {
        Future<String> future = asyncService.asyncTaskD("Mr.nobody");
        LOGGER.info("future result:" + future.get());
        LOGGER.info("主线程执行...");
    }
}










