Async 异步任务

03 SpringBoot 评论

1. 简介

当我们在使用Springboot进行开发时，可能会遇到一些异步任务的场景，如果每次执行这些异步任务都去新建一个异步线程来执行的话，那么代码就太冗余了。Springboot提供了@Async注解，能够轻松地对这些异步任务进行执行

失效条件

  1. 异步方法使用static修饰

  2. 调用方法和异步方法在同一个类中

2. 代码工程

基于springboot实现@Async异步任务

  1. 添加依赖 
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    <?xml version="1.0" encoding="UTF-8"?>
     <project xmlns="http://maven.apache.org/POM/4.0.0"
              xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
              xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
         <parent>
             <artifactId>springboot-demo</artifactId>
             <groupId>com.ygb</groupId>
             <version>1.0-SNAPSHOT</version>
         </parent>
         <modelVersion>4.0.0</modelVersion>
         <artifactId>async</artifactId>

         <properties>
             <maven.compiler.source>8</maven.compiler.source>
             <maven.compiler.target>8</maven.compiler.target>
         </properties>
         <dependencies>
             <dependency>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-starter-web</artifactId>
             </dependency>

             <dependency>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-autoconfigure</artifactId>
             </dependency>
             <dependency>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-starter-test</artifactId>
                 <scope>test</scope>
             </dependency>
             <dependency>
                 <groupId>org.projectlombok</groupId>
                 <artifactId>lombok</artifactId>
             </dependency>
         </dependencies>
     </project>

  1. Service

    • thenApply: 处理上一阶段计算结果

    • thenCompose: 整合两个计算结果

      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      @Service
      @Slf4j
      public class FirstAsyncService {
          @Async
          public CompletableFuture<String> asyncGetData() throws InterruptedException {
             log.info("Execute method asynchronously " + Thread.currentThread().getName());
              Thread.sleep(4000);
              return new AsyncResult<>(super.getClass().getSimpleName() + " response !!! ").completable();
          }
      }
      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      @Service
      @Slf4j
      public class SecondAsyncService {
          @Async
          public CompletableFuture<String> asyncGetData() throws InterruptedException {
              log.info("Execute method asynchronously " + Thread.currentThread()
                      .getName());
              Thread.sleep(4000);
              return new AsyncResult<>(super.getClass().getSimpleName() + " response !!! ").completable();
          }
      }

      整合两个计算结果

      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      20
      21
      22
      23
      24
      25
      26
      27
      28
      29
      30
      31
      32
      33
      34
      35
      36
      37
      @Service
      @Slf4j
      public class NotifyService {
          public void noAsync() {
             log.info("Execute method asynchronously. " + Thread.currentThread().getName());
          }
          public void withAsync() {
              log.info("Execute method asynchronously. " + Thread.currentThread().getName());
          }
          @Async("threadPoolTaskExecutor")
          public void mockerror() {
              int ss=12/0;
          }
          @Async
          public Future<String> asyncMethodWithReturnType() {
              log.info("Execute method asynchronously - " + Thread.currentThread().getName());
              try {
                  Thread.sleep(5000);
                  return new AsyncResult<String>("hello world !!!!");
              } catch (InterruptedException e) {
                  e.printStackTrace();
              }
              return null;
          }
          @Autowired
          private FirstAsyncService fisrtService;
          @Autowired
          private SecondAsyncService secondService;

          public CompletableFuture<String> asyncMergeServicesResponse() throws InterruptedException {
              CompletableFuture<String> fisrtServiceResponse = fisrtService.asyncGetData();
              CompletableFuture<String> secondServiceResponse = secondService.asyncGetData();

              // Merge responses from FirstAsyncService and SecondAsyncService
              return fisrtServiceResponse.thenCompose(fisrtServiceValue -> secondServiceResponse.thenApply(secondServiceValue -> fisrtServiceValue + secondServiceValue));
          }
      }

  2. config,使用@EnableAsync开启

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    @Configuration
    @EnableAsync
    public class SpringAsyncConfig implements AsyncConfigurer {

        @Bean(name = "threadPoolTaskExecutor")
        public Executor threadPoolTaskExecutor() {
            return new ThreadPoolTaskExecutor();
        }

        @Override
        public Executor getAsyncExecutor() {
            ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
            threadPoolTaskExecutor.initialize();
            return threadPoolTaskExecutor;
        }

        @Override
        public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
            return new CustomAsyncExceptionHandler();
        }
    }

  3. 自定义异常处理

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    public class CustomAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {

        @Override
        public void handleUncaughtException(
                Throwable throwable, Method method, Object... obj) {
     
            System.out.println("Exception message - " + throwable.getMessage());
            System.out.println("Method name - " + method.getName());
            for (Object param : obj) {
                System.out.println("Parameter value - " + param);
            }
        }
        
    }

  4. 启动类

    1
    2
    3
    4
    5
    6
    7
    @SpringBootApplication
    public class DemoApplication {

        public static void main(String[] args) {
            SpringApplication.run(DemoApplication.class, args);
        }
    }

  5. 测试类

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    @RunWith(SpringRunner.class)
    @SpringBootTest(classes = DemoApplication.class)
    @Slf4j
    public class DemoTests {
        @Autowired
        NotifyService notifyService;

        @Before
        public void before()  {
            log.info("init some data");
        }
        @After
        public void after(){
            log.info("clean some data");
        }
        @Test
        public void execute() throws ExecutionException, InterruptedException {
            log.info("your method test Code");
            log.info("Invoking an asynchronous method. " + Thread.currentThread().getName());
            notifyService.noAsync();
            notifyService.withAsync();

        }
        @Test
        public void mockerror() throws ExecutionException, InterruptedException {
            notifyService.mockerror();
        }
        @Test
        public void testAsyncAnnotationForMethodsWithReturnType()
                throws InterruptedException, ExecutionException {
            log.info("Invoking an asynchronous method. " + Thread.currentThread().getName());
            Future<String> future = notifyService.asyncMethodWithReturnType();

            while (true) {
                if (future.isDone()) {
                    log.info("Result from asynchronous process - " + future.get());
                    break;
                }
                log.info("Continue doing something else. ");
                Thread.sleep(1000);
            }
        }

        @Test
        public void testAsyncAnnotationForMergedServicesResponse() throws InterruptedException, ExecutionException {
           log.info("Invoking an asynchronous method. " + Thread.currentThread().getName());
            CompletableFuture<String> completableFuture = notifyService.asyncMergeServicesResponse();

            while (true) {
                if (completableFuture.isDone()) {
                   log.info("Result from asynchronous process - " + completableFuture.get());
                    break;
                }
                log.info("Continue doing something else. ");
                Thread.sleep(1000);
            }
        }
    }
    1
    2
    3
    4
    5
    6
    2024-11-15 14:17:24.723  INFO 1066 --- [           main] com.et.async.DemoTests                   : init some data
    2024-11-15 14:17:24.724  INFO 1066 --- [           main] com.et.async.DemoTests                   : your method test Code
    2024-11-15 14:17:24.724  INFO 1066 --- [           main] com.et.async.DemoTests                   : Invoking an asynchronous method. main
    2024-11-15 14:17:24.730  INFO 1066 --- [           main] com.et.async.service.NotifyService       : Execute method asynchronously. main
    2024-11-15 14:17:24.730  INFO 1066 --- [           main] com.et.async.service.NotifyService       : Execute method asynchronously. main
    2024-11-15 14:17:24.732  INFO 1066 --- [           main] com.et.async.DemoTests                   : clean some data

小余哥xiaoyouge

分享一些技术教程与实践，以及日常踩坑记录。