Spring 异步调用，多线程，一行代码实现 Spring

发表于

Spring 异步调用，多线程

  • 概述
  • 快速入门
  • 异步回调
  • 异步异常处理
  • 自定义执行器

1、概述

在日常开发中，我们的逻辑都是同步调用，顺序执行。但是在某些情况下我们希望异步调用，将主线程和部分逻辑分开，以达到程序更快速的执行，提升性能。例如，高并发的接口，用户操作日志等。

异步调用，对应的是同步调用。

  • 同步调用：指程序按照 定义顺序 依次执行，每一行程序都必须等待上一行程序执行完成之后才能执行；
  • 异步调用：指程序在顺序执行时，不等异步调用返回执行结果，就执行后面的程序。

考虑到异步的可靠性，我们一般会考虑引入消息队列，例如： RabbitMQ、RocketMQ、Kafka 等等。 但是在一些时候，我们不需要如此高的可靠性，可以使用进程内的队列或线程池。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
复制代码public static void main(String[] args) {
        // 创建线程池。这里只是临时测试，开发规范...
        ExecutorService executor = Executors.newFixedThreadPool(10);

        // 提交任务到线程池中执行。
        executor.submit(new Runnable() {

            @Override
            public void run() {
                System.out.println("听说我被异步调用了");
            }

        });
    }

在进程内的队列或者线程池，相对不可靠的原因是，队列和线程池中的任务仅仅存储在内存中，如何JVM进程被异常关闭，将会导致丢失，未被执行。

而分布式消息队列，异步调用会以一个消息的形式，存储在消息服务器上，所以即使JVM进程被异常中断，消息依然在消息服务队列的服务器上

所以使用进程内的队列或者线程池来实现异步调用的话，一定要尽可能的保证JVM进程的优雅关闭，保证他们在关闭前被执行完。

在 Spring Framework 的 Spring Task 模块，提供了 @Async 注解，可以添加在方法上，自动实现该方法的异步调用

简单来说，我们可以像使用 @Transactional声明式事务，使用SpringTask提供的@Async注解，声明式异步。而在实现原理上，也是基于 Spring AOP 拦截，实现异步提交该操作到线程池中，达到异步调用的目的。

2、快速入门

2.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
复制代码<?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>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.1.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>lab-29-async-demo</artifactId>

    <dependencies>
        <!-- 引入 Spring Boot 依赖 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <!-- 方便等会写单元测试 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

</project>

因为 Spring Task 是 Spring Framework 的模块，所以在我们引入 spring-boot-web 依赖后，无需特别引入它。

2.2 Application

创建Application类，添加@EnableAsync 开启 @Async 的支持

1
2
3
4
5
6
7
8
复制代码@SpringBootApplication
@EnableAsync // 开启 @Async 的支持
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}
  • 在类上添加 @EnableAsync 注解，启用异步功能。

2.3 DemoService

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
复制代码package cn.iocoder.springboot.lab29.asynctask.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

@Service
public class DemoService {

    private Logger logger = LoggerFactory.getLogger(getClass());
    
    public Integer execute01() {
        logger.info("[execute01]");
        sleep(10);
        return 1;
    }

    public Integer execute02() {
        logger.info("[execute02]");
        sleep(5);
        return 2;
    }

    private static void sleep(int seconds) {
        try {
            Thread.sleep(seconds * 1000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    @Async
    public Integer zhaoDaoNvPengYou(Integer a, Integer b) {
        throw new RuntimeException("程序员不需要女朋友");
    }

}
  • 定义 execute01和 execute02方法，分别模拟sleep 10秒和5秒。
  • 同时在方法中，使用logger打印日志，方便我们看到每个方法的执行时间，和执行的线程

2.4 同步调用测试

编写DemoServiceTest测试类，添加#task01()方法，同步调用上述方法，代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
复制代码@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class DemoServiceTest {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private DemoService demoService;

    @Test
    public void task01() {
        long now = System.currentTimeMillis();
        logger.info("[task01][开始执行]");

        demoService.execute01();
        demoService.execute02();

        logger.info("[task01][结束执行，消耗时长 {} 毫秒]", System.currentTimeMillis() - now);
    }
}

运行单元测试，打印日志如下：

1
2
3
4
复制代码2020-06-02 09:16:03.391  INFO 3108 --- [      main] c.i.s.l.a.service.DemoServiceTest        : [task01][开始执行]
2020-06-02 09:16:03.402  INFO 3108 --- [      main] c.i.s.l.asynctask.service.DemoService    : [execute01]
2020-06-02 09:16:13.403  INFO 3108 --- [      main] c.i.s.l.asynctask.service.DemoService    : [execute02]
2020-06-02 09:16:18.403  INFO 3108 --- [      main] c.i.s.l.a.service.DemoServiceTest        : [task01][结束执行，消耗时长 15012 毫秒]
  • 两个方法都按顺序执行，执行时间15秒。
  • 都在主线程执行。

2.5 异步调用测试

修改DemoServiceTest ，增加 execute01Async()和execute02Async()异步调用方法，代码：

1
2
3
4
5
6
7
8
9
复制代码	@Async
    public Integer execute01Async() {
        return this.execute01();
    }

    @Async
    public Integer execute02Async() {
        return this.execute02();
    }
  • 在execute01Async() 和 execute01Async()上，添加@Async实现异步调用

修改DemoServiceTest类， 编写 #task02() 方法，异步调用上述的两个方法。

1
2
3
4
5
6
7
8
9
10
复制代码	@Test
    public void task02() {
        long now = System.currentTimeMillis();
        logger.info("[task02][开始执行]");

        demoService.execute01Async();
        demoService.execute02Async();

        logger.info("[task02][结束执行，消耗时长 {} 毫秒]", System.currentTimeMillis() - now);
    }

打印日志：

1
2
3
复制代码2020-06-02 10:57:41.643  INFO 14416 --- [main] c.i.s.l.a.service.DemoServiceTest        : [task02][开始执行]
2020-06-02 10:57:41.675  INFO 14416 --- [main] o.s.s.concurrent.ThreadPoolTaskExecutor  : Initializing ExecutorService 'applicationTaskExecutor'
2020-06-02 10:57:41.682  INFO 14416 --- [main] c.i.s.l.a.service.DemoServiceTest        : [task02][结束执行，消耗时长 39 毫秒]
  • DemoService 的两个方法，异步执行，所以主线程只消耗 39毫秒左右。注意，实际这两个方法，并没有执行完成。
  • DemoService 的两个方法，都在异步线程池中执行。

2.6 等待异步调用完成测试

在上面的**【2.5 异步调用测试】**异步调用中，两个方法只是异步调用，方法没有执行完。在一些业务场景中，我们达到异步调用效果，同时主线程有返回结果，就需要主线程阻塞等待异步调用的结果。

修改DemoService，添加execute01AsyncWithFuture()和execute01AsyncWithFuture()异步调用，并返回 Future 对象 。代码：

1
2
3
4
5
6
7
8
9
复制代码	@Async
    public Future<Integer> execute01AsyncWithFuture() {
        return AsyncResult.forValue(this.execute01());
    }

    @Async
    public Future<Integer> execute02AsyncWithFuture() {
        return AsyncResult.forValue(this.execute02());
    }
  • 在这里两个异步方法中，添加了AsyncResult.forValue(this.execute02());,返回带有执行结果的Future对象

修改DemoServiceTest类， 编写 #task02() 方法，异步调用上述的两个方法,并阻塞线程等待异步调用返回结果

代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
复制代码	@Test
    public void task03() throws ExecutionException, InterruptedException {
        long now = System.currentTimeMillis();
        logger.info("[task03][开始执行]");

        // 执行任务
        Future<Integer> execute01Result = demoService.execute01AsyncWithFuture();
        Future<Integer> execute02Result = demoService.execute02AsyncWithFuture();
        // 阻塞等待结果
        execute01Result.get();
        execute02Result.get();

        logger.info("[task03][结束执行，消耗时长 {} 毫秒]", System.currentTimeMillis() - now);
    }
  • 异步调用两个方法，并返回对应Future对象。这两个的异步调用逻辑，可以并行执行。
  • Future对象的get()方法，效果：阻塞线程等待返回结果。

打印日志：

1
2
3
4
5
复制代码2020-06-02 13:56:43.955  INFO 7828 --- [ main] c.i.s.l.a.service.DemoServiceTest        : [task03][开始执行]
2020-06-02 13:56:43.987  INFO 7828 --- [ main] o.s.s.concurrent.ThreadPoolTaskExecutor  : Initializing ExecutorService 'applicationTaskExecutor'
2020-06-02 13:56:44.008  INFO 7828 --- [ task-1] c.i.s.l.asynctask.service.DemoService    : [execute01]
2020-06-02 13:56:44.008  INFO 7828 --- [ task-2] c.i.s.l.asynctask.service.DemoService    : [execute02]
2020-06-02 13:56:54.008  INFO 7828 --- [ main] c.i.s.l.a.service.DemoServiceTest        : [task03][结束执行，消耗时长 10053 毫秒]
  • 两个异步调用方法，分别由线程池 task-1和task-2 同时执行。 因为主线程阻塞等待执行结果 ，执行时间10秒，当同时有多个异步调用，线程阻塞等待，执行时间由消耗最长的异步调用逻辑所决定。

2.7 应用配置文件

在application中，添加spring Task配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
复制代码spring:
  task:
    # Spring 执行器配置，对应 TaskExecutionProperties 配置类。对于 Spring 异步任务，会使用该执行器。
    execution:
      thread-name-prefix: task- # 线程池的线程名的前缀。默认为 task- ，建议根据自己应用来设置
      pool: # 线程池相关
        core-size: 8 # 核心线程数，线程池创建时候初始化的线程数。默认为 8 。
        max-size: 20 # 最大线程数，线程池最大的线程数，只有在缓冲队列满了之后，才会申请超过核心线程数的线程。默认为 Integer.MAX_VALUE
        keep-alive: 60s # 允许线程的空闲时间，当超过了核心线程之外的线程，在空闲时间到达之后会被销毁。默认为 60 秒
        queue-capacity: 200 # 缓冲队列大小，用来缓冲执行任务的队列的大小。默认为 Integer.MAX_VALUE 。
        allow-core-thread-timeout: true # 是否允许核心线程超时，即开启线程池的动态增长和缩小。默认为 true 。
      shutdown:
        await-termination: true # 应用关闭时，是否等待定时任务执行完成。默认为 false ，建议设置为 true
        await-termination-period: 60 # 等待任务完成的最大时长，单位为秒。默认为 0 ，根据自己应用来设置
  • Spring 本身依赖了Spring Task
  • 在spring.task.execution配置项， Spring Task 调度任务的配置 ，对应TaskExecutionProperties配置类
  • Spring Boot TaskExecutionAutoConfiguration 自动化配置类， 实现了Spring Task 自动配置，创建了ThreadPoolTaskExecutor基于线程池的任务执行器，实际上ThreadPoolTaskExecutor就是ThreadPoolExecutor的分装，主要增加执行任务，并返回 ListenableFuture 对象功能。

之前说的异步的可靠性，要优雅的关闭进程。spring.task.execution.shutdown配置关闭，是为了实现Spring Task的优雅关闭。异步任务在执行过程中，如果应用开始关闭，异步任务需要使用的Bean被销毁，例如：需要访问数据库连接池，这时候异步任务还在执行中，一旦需要访问数据库，但是没有对应的Bean将会导致报错。

  • 通过配置await-termination: true，实现在应用关闭时，等待异步任务执行完成。这样在应用关闭时，Spring 会等待 ThreadPoolTaskExecutor执行完任务，再销毁Bean。
  • 应用关闭时，在某些业务场景下我们不可能让Spring一直等待，异步任务的完成。通过配置await-termination-period: 60,设置Spring最大等待时间，时间一到将不再等待异步任务完成。

3、异步回调

业务场景中，执行完异步任务，可能需回调。下面介绍异步执行完成后，实现自定义回调。

3.1、AsyncResult 源码解释

在 2.6 等待异步调用完成 中，我们看到的 AsyncResult类 表示异步结果。返回结果分为两种情况：

  • 执行成功时，调用AsyncResult#forValue(V value) 静态方法，返回成功的 ListenableFuture对象，

源码:

1
2
3
4
5
6
7
8
9
复制代码	/**
	 * Create a new async result which exposes the given value from {@link Future#get()}.
	 * @param value the value to expose
	 * @since 4.2
	 * @see Future#get()
	 */
	public static <V> ListenableFuture<V> forValue(V value) {
		return new AsyncResult<>(value, null);
	}
  • 执行异常时，调用 AsyncResult#forExecutionException(Throwable ex) 静态方法，返回异常的 ListenableFuture 对象。源码：
1
2
3
4
5
6
7
8
9
10
11
复制代码	/**
	 * Create a new async result which exposes the given exception as an
	 * {@link ExecutionException} from {@link Future#get()}.
	 * @param ex the exception to expose (either an pre-built {@link ExecutionException}
	 * or a cause to be wrapped in an {@link ExecutionException})
	 * @since 4.2
	 * @see ExecutionException
	 */
	public static <V> ListenableFuture<V> forExecutionException(Throwable ex) {
		return new AsyncResult<>(null, ex);
	}

AsyncResult 同时也实现了 ListenableFuture接口，提供异步执行结果回调处理。

1
复制代码public class AsyncResult<V> implements ListenableFuture<V>

ListenableFuture接口，源码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
复制代码public interface ListenableFuture<T> extends Future<T> {

    // 添加回调方法，统一处理成功和异常的情况。
	void addCallback(ListenableFutureCallback<? super T> callback);

	// 添加成功和失败的回调方法，分别处理成功和异常的情况。
	void addCallback(SuccessCallback<? super T> successCallback, FailureCallback failureCallback);


	// 将 ListenableFuture 转换成 JDK8 提供的 CompletableFuture 。
    // 这样，后续我们可以使用 ListenableFuture 来设置回调
	default CompletableFuture<T> completable() {
		CompletableFuture<T> completable = new DelegatingCompletableFuture<>(this);
		addCallback(completable::complete, completable::completeExceptionally);
		return completable;
	}

}

ListenableFuture继承了Future，所以AsyncResult 也实现了Future的接口，源码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
复制代码public interface Future<V> {

    // 如果任务还没开始，执行 cancel(...) 方法将返回 false；
    // 如果任务已经启动，执行 cancel(true) 方法将以中断执行此任务线程的方式来试图停止任务，如果停止成功，返回 true ；
    // 当任务已经启动，执行 cancel(false) 方法将不会对正在执行的任务线程产生影响(让线程正常执行到完成)，此时返回 false ；
    // 当任务已经完成，执行 cancel(...) 方法将返回 false 。
    // mayInterruptRunning 参数表示是否中断执行中的线程。
    boolean cancel(boolean mayInterruptIfRunning);

    // 如果任务完成前被取消，则返回 true 。
    boolean isCancelled();

    // 如果任务执行结束，无论是正常结束或是中途取消还是发生异常，都返回 true 。
    boolean isDone();

    // 获取异步执行的结果，如果没有结果可用，此方法会阻塞直到异步计算完成。
    V get() throws InterruptedException, ExecutionException;
    
	// 获取异步执行结果，如果没有结果可用，此方法会阻塞，但是会有时间限制，如果阻塞时间超过设定的 timeout 时间，该方法将抛出异常。
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

AsyncResult 中对addCallback(...)方法回调的实现,源码：

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
复制代码	@Override
	public void addCallback(ListenableFutureCallback<? super V> callback) {
		addCallback(callback, callback);
	}

	@Override
	public void addCallback(SuccessCallback<? super V> successCallback, FailureCallback failureCallback) {
		try {
			if (this.executionException != null) { // 《1》
				failureCallback.onFailure(exposedException(this.executionException));
			}
			else { // 《2》
				successCallback.onSuccess(this.value);
			}
		}
		catch (Throwable ex) { // 《3》
			// Ignore
		}
	}

// 从 ExecutionException 中，获得原始异常。
private static Throwable exposedException(Throwable original) {
    if (original instanceof ExecutionException) {
        Throwable cause = original.getCause();
        if (cause != null) {
            return cause;
        }
    }
    return original;
}
  • 从 ListenableFutureCallback 知道 ，ListenableFutureCallback 接口同时继承了 SuccessCallback、FailureCallback接口
1
复制代码public interface ListenableFutureCallback<T> extends SuccessCallback<T>, FailureCallback
  • 《1》，如果是异常处理结果调用 failureCallback回调
  • 《2》，如果是成功处理结果调用successCallback回调
  • 《3》，如果回调逻辑发生异常，直接忽略。假设多个回调，其中一个出现议程，不会影响其他的回调。

实际上，AsyncResult 是作为异步执行的结果。既然是结果，执行就已经完成。所以，在我们调用 #addCallback(...) 接口方法来添加回调时，必然直接使用回调处理执行的结果。

AsyncResult 对 Future 定义的所有方法，实现代码如下：

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
复制代码// AsyncResult.java

@Override
public boolean cancel(boolean mayInterruptIfRunning) {
    return false; // 因为是 AsyncResult 是执行结果，所以直接返回 false 表示取消失败。
}

@Override
public boolean isCancelled() {
    return false; // 因为是 AsyncResult 是执行结果，所以直接返回 false 表示未取消。
}

@Override
public boolean isDone() {
    return true; // 因为是 AsyncResult 是执行结果，所以直接返回 true 表示已完成。
}

@Override
@Nullable
public V get() throws ExecutionException {
    // 如果发生异常，则抛出该异常。
    if (this.executionException != null) {
        throw (this.executionException instanceof ExecutionException ?
                (ExecutionException) this.executionException :
                new ExecutionException(this.executionException));
    }
    // 如果执行成功，则返回该 value 结果
    return this.value;
}

@Override
@Nullable
public V get(long timeout, TimeUnit unit) throws ExecutionException {
    return get();
}

3.2 ListenableFutureTask

在我们调用使用 @Async 注解的方法时，如果方法返回的类型是 ListenableFuture 的情况下，实际方法返回的是 ListenableFutureTask 对象。

ListenableFutureTask 类，也实现 ListenableFuture 接口，继承 FutureTask 类，ListenableFuture 的 FutureTask 实现类。

ListenableFutureTask 对 ListenableFuture 定义的 #addCallback(...) 方法，实现源码如下：

1
2
3
4
5
6
7
8
9
10
11
12
复制代码private final ListenableFutureCallbackRegistry<T> callbacks = new ListenableFutureCallbackRegistry<T>();

@Override
public void addCallback(ListenableFutureCallback<? super T> callback) {
    this.callbacks.addCallback(callback);
}

@Override
public void addCallback(SuccessCallback<? super T> successCallback, FailureCallback failureCallback) {
    this.callbacks.addSuccessCallback(successCallback);
    this.callbacks.addFailureCallback(failureCallback);
}
  • 可以看到在ListenableFutureTask中，暂存回调到ListenableFutureCallbackRegistry中

ListenableFutureTask 对 FutureTask 已实现的 #done() 方法，进行重写。实现源码如下：

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
复制代码@Override
	protected void done() {
		Throwable cause;
		try {
            // 获得执行结果
			T result = get();
            // 执行成功，执行成功的回调
			this.callbacks.success(result);
			return;
		}
		catch (InterruptedException ex) { // 如果有中断异常 InterruptedException异常，则打断当前线程，直接返回
			Thread.currentThread().interrupt();
			return;
		}
		catch (ExecutionException ex) { // 如果有 ExecutionException 异常，获取真实异常，并设置到cause中
			cause = ex.getCause();
			if (cause == null) {
				cause = ex;
			}
		}
		catch (Throwable ex) {
			cause = ex; // 设置异常到 cause 中
		}
         // 执行异常，执行异常的回调
		this.callbacks.failure(cause);
	}

3.3 具体示例

修改 DemoService 的代码，增加 #execute02() 的异步调用，并返回 ListenableFuture 对象。代码如下:

1
2
3
4
5
6
7
8
9
复制代码@Async
    public ListenableFuture<Integer> execute01AsyncWithListenableFuture() {
        try {
            //int i = 1 / 0;
            return AsyncResult.forValue(this.execute02());
        } catch (Throwable ex) {
            return AsyncResult.forExecutionException(ex);
        }
    }
  • 根据执行的结果，包装出成功还是异常的 AsyncResult 对象。

DemoServiceTest 测试类，编写 #task04() 方法，异步调用上述的方法，在塞等待执行完成的同时，添加相应的回调 Callback 方法。代码：

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
复制代码@Test
    public void task04() throws ExecutionException, InterruptedException {
        long now = System.currentTimeMillis();
        logger.info("[task04][开始执行]");

        // <1> 执行任务
        ListenableFuture<Integer> execute01Result = demoService.execute01AsyncWithListenableFuture();
        logger.info("[task04][execute01Result 的类型是：({})]",execute01Result.getClass().getSimpleName());
        execute01Result.addCallback(new SuccessCallback<Integer>() { // <2.1> 增加成功的回调

            @Override
            public void onSuccess(Integer result) {
                logger.info("[onSuccess][result: {}]", result);
            }

        }, new FailureCallback() { // <2.1> 增加失败的回调

            @Override
            public void onFailure(Throwable ex) {
                logger.info("[onFailure][发生异常]", ex);
            }

        });
        execute01Result.addCallback(new ListenableFutureCallback<Integer>() { // <2.2> 增加成功和失败的统一回调

            @Override
            public void onSuccess(Integer result) {
                logger.info("[onSuccess][result: {}]", result);
            }

            @Override
            public void onFailure(Throwable ex) {
                logger.info("[onFailure][发生异常]", ex);
            }

        });
        // <3> 阻塞等待结果
        execute01Result.get();

        logger.info("[task04][结束执行，消耗时长 {} 毫秒]", System.currentTimeMillis() - now);
    }
  • <1> 处，调用 DemoService#execute01AsyncWithListenableFuture() 方法，异步调用该方法，并返回 ListenableFutureTask 对象。这里，我们看下打印的日志。
1
复制代码2020-06-08 14:13:16.738  INFO 5060 --- [  main] c.i.s.l.a.service.DemoServiceTest : [task04][execute01Result 的类型是：(ListenableFutureTask)]
  • <2.1> 处，增加成功的回调和失败的回调。
  • <2.2> 处，增加成功和失败的统一回调。
  • <3> 处，阻塞等待结果。执行完成后，我们会看到回调被执行，打印日志如下：
1
2
3
复制代码2020-06-08 14:13:21.752  INFO 5060 --- [   main] c.i.s.l.a.service.DemoServiceTest   : [task04][结束执行，消耗时长 5057 毫秒]
2020-06-08 14:13:21.752  INFO 5060 --- [ task-1] c.i.s.l.a.service.DemoServiceTest   : [onSuccess][result: 2]
2020-06-08 14:13:21.752  INFO 5060 --- [ task-1] c.i.s.l.a.service.DemoServiceTest   : [onSuccess][result: 2]
  1. 异步异常处理器

通过实现 AsyncUncaughtExceptionHandler 接口，达到对异步调用的异常的统一处理。

创建 GlobalAsyncExceptionHandler 类，全局统一的异步调用异常的处理器。代码：

1
2
3
4
5
6
7
8
9
10
11
12
复制代码@Component
public class GlobalAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    public void handleUncaughtException(Throwable ex, Method method, Object... params) {
        logger.error("[handleUncaughtException][method({}) params({}) 发生异常]",
                method, params, ex);
    }

}
  • 类上，我们添加了 @Component 注解，考虑到胖友可能会注入一些 Spring Bean 到属性中。
  • 实现 #handleUncaughtException(Throwable ex, Method method, Object... params) 方法，打印异常日志。

注意，AsyncUncaughtExceptionHandler 只能拦截返回类型非 Future 的异步调用方法。通过看 AsyncExecutionAspectSupport#handleError(Throwable ex, Method method, Object... params) 的源码，可以很容易得到这个结论，代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
复制代码// AsyncExecutionAspectSupport.java

protected void handleError(Throwable ex, Method method, Object... params) throws Exception {
    // 重点！！！如果返回类型是 Future ，则直接抛出该异常。
    if (Future.class.isAssignableFrom(method.getReturnType())) {
        ReflectionUtils.rethrowException(ex);
    } else {
        // 否则，交给 AsyncUncaughtExceptionHandler 来处理。
        // Could not transmit the exception to the caller with default executor
        try {
            this.exceptionHandler.obtain().handleUncaughtException(ex, method, params);
        } catch (Throwable ex2) {
            logger.warn("Exception handler for async method '" + method.toGenericString() +
                    "' threw unexpected exception itself", ex2);
        }
    }
}
  • 对了，AsyncExecutionAspectSupport 是 AsyncExecutionInterceptor 的父类哟。所以哟，返回类型为 Future 的异步调用方法，需要通过「3. 异步回调」来处理。

4.2 AsyncConfig

创建 AsyncConfig 类，配置异常处理器。代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
复制代码@Configuration
@EnableAsync // 开启 @Async 的支持
public class AsyncConfig implements AsyncConfigurer {

    @Autowired
    private GlobalAsyncExceptionHandler exceptionHandler;

    @Override
    public Executor getAsyncExecutor() {
        return null;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return exceptionHandler;
    }

}
  • 在类上添加 @EnableAsync 注解，启用异步功能。这样「2. Application」 的 @EnableAsync 注解，也就可以去掉了。
  • 实现 AsyncConfigurer 接口，实现异步相关的全局配置。 此时此刻，胖友有没想到 SpringMVC 的 WebMvcConfigurer 接口。
  • 实现 #getAsyncUncaughtExceptionHandler() 方法，返回我们定义的 GlobalAsyncExceptionHandler 对象。
  • 实现 #getAsyncExecutor() 方法，返回 Spring Task 异步任务的默认执行器。这里，我们返回了 null ，并未定义默认执行器。所以最终会使用 TaskExecutionAutoConfiguration 自动化配置类创建出来的 ThreadPoolTaskExecutor 任务执行器，作为默认执行器。

4.3 DemoService

DemoService 类，增加 #zhaoDaoNvPengYou(...) 的异步调用。代码如下：

1
2
3
4
复制代码@Async
public Integer zhaoDaoNvPengYou(Integer a, Integer b) {
    throw new RuntimeException("异步全局异常");
}

4.4 简单测试

1
2
3
4
5
6
7
复制代码 @Test
    public void testZhaoDaoNvPengYou() throws InterruptedException {
        demoService.zhaoDaoNvPengYou(1, 2);

        // sleep 1 秒，保证异步调用的执行
        Thread.sleep(1000);
    }

运行单元测试，执行日志如下：

1
2
3
复制代码2020-06-08 15:26:35.120 ERROR 11388 --- [         task-1] .i.s.l.a.c.a.GlobalAsyncExceptionHandler : [handleUncaughtException][method(public java.lang.Integer cn.iocoder.springboot.lab29.asynctask.service.DemoService.zhaoDaoNvPengYou(java.lang.Integer,java.lang.Integer)) params([1, 2]) 发生异常]

java.lang.RuntimeException: 异步全局异常
  1. 自定义执行器

在 上面 中，我们使用 Spring Boot TaskExecutionAutoConfiguration 自动化配置类，实现自动配置 ThreadPoolTaskExecutor 任务执行器。

本小节，我们希望两个自定义 ThreadPoolTaskExecutor 任务执行器，实现不同方法，分别使用这两个 ThreadPoolTaskExecutor 任务执行器。

5.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
复制代码<?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>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.1.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>lab-29-async-demo</artifactId>

    <dependencies>
        <!-- 引入 Spring Boot 依赖 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <!-- 方便等会写单元测试 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

</project>
  • 和 上面引入依赖 一致。

5.2 应用配置文件

在 application.yml 中，添加 Spring Task 定时任务的配置，如下：

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
复制代码spring:
  task:
    # Spring 执行器配置，对应 TaskExecutionProperties 配置类。对于 Spring 异步任务，会使用该执行器。
    execution-one:
      thread-name-prefix: task-one- # 线程池的线程名的前缀。默认为 task- ，建议根据自己应用来设置
      pool: # 线程池相关
        core-size: 8 # 核心线程数，线程池创建时候初始化的线程数。默认为 8 。
        max-size: 20 # 最大线程数，线程池最大的线程数，只有在缓冲队列满了之后，才会申请超过核心线程数的线程。默认为 Integer.MAX_VALUE
        keep-alive: 60s # 允许线程的空闲时间，当超过了核心线程之外的线程，在空闲时间到达之后会被销毁。默认为 60 秒
        queue-capacity: 200 # 缓冲队列大小，用来缓冲执行任务的队列的大小。默认为 Integer.MAX_VALUE 。
        allow-core-thread-timeout: true # 是否允许核心线程超时，即开启线程池的动态增长和缩小。默认为 true 。
      shutdown:
        await-termination: true # 应用关闭时，是否等待定时任务执行完成。默认为 false ，建议设置为 true
        await-termination-period: 60 # 等待任务完成的最大时长，单位为秒。默认为 0 ，根据自己应用来设置
    # Spring 执行器配置，对应 TaskExecutionProperties 配置类。对于 Spring 异步任务，会使用该执行器。
    execution-two:
      thread-name-prefix: task-two- # 线程池的线程名的前缀。默认为 task- ，建议根据自己应用来设置
      pool: # 线程池相关
        core-size: 8 # 核心线程数，线程池创建时候初始化的线程数。默认为 8 。
        max-size: 20 # 最大线程数，线程池最大的线程数，只有在缓冲队列满了之后，才会申请超过核心线程数的线程。默认为 Integer.MAX_VALUE
        keep-alive: 60s # 允许线程的空闲时间，当超过了核心线程之外的线程，在空闲时间到达之后会被销毁。默认为 60 秒
        queue-capacity: 200 # 缓冲队列大小，用来缓冲执行任务的队列的大小。默认为 Integer.MAX_VALUE 。
        allow-core-thread-timeout: true # 是否允许核心线程超时，即开启线程池的动态增长和缩小。默认为 true 。
      shutdown:
        await-termination: true # 应用关闭时，是否等待定时任务执行完成。默认为 false ，建议设置为 true
        await-termination-period: 60 # 等待任务完成的最大时长，单位为秒。默认为 0 ，根据自己应用来设置
  • 在 spring.task 配置项下，我们新增了 execution-one 和 execution-two 两个执行器的配置。在格式上，我们保持和在「2.7 应用配置文件」看到的 spring.task.exeuction 一致，方便我们后续复用 TaskExecutionProperties 属性配置类来映射。

5.3 AsyncConfig

创建 AsyncConfig 类，配置两个执行器。代码如下：

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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
复制代码@Configuration
@EnableAsync // 开启 @Async 的支持
public class AsyncConfig
{

    public static final String EXECUTOR_ONE_BEAN_NAME = "executor-one";
    public static final String EXECUTOR_TWO_BEAN_NAME = "executor-two";

    @Configuration
    public static class ExecutorOneConfiguration
    {

        @Bean(name = EXECUTOR_ONE_BEAN_NAME + "-properties")
        @Primary
        @ConfigurationProperties(prefix = "spring.task.execution-one")
        // 读取 spring.task.execution-one 配置到 TaskExecutionProperties 对象
        public TaskExecutionProperties taskExecutionProperties()
        {
            return new TaskExecutionProperties();
        }

        @Bean(name = EXECUTOR_ONE_BEAN_NAME)
        public ThreadPoolTaskExecutor threadPoolTaskExecutor()
        {
            // 创建 TaskExecutorBuilder 对象
            TaskExecutorBuilder builder = createTskExecutorBuilder(this.taskExecutionProperties());
            // 创建 ThreadPoolTaskExecutor 对象
            return builder.build();
        }

    }

    @Configuration
    public static class ExecutorTwoConfiguration
    {

        @Bean(name = EXECUTOR_TWO_BEAN_NAME + "-properties")
        @ConfigurationProperties(prefix = "spring.task.execution-two")
        // 读取 spring.task.execution-two 配置到 TaskExecutionProperties 对象
        public TaskExecutionProperties taskExecutionProperties()
        {
            return new TaskExecutionProperties();
        }

        @Bean(name = EXECUTOR_TWO_BEAN_NAME)
        public ThreadPoolTaskExecutor threadPoolTaskExecutor()
        {
            // 创建 TaskExecutorBuilder 对象
            TaskExecutorBuilder builder = createTskExecutorBuilder(this.taskExecutionProperties());
            // 创建 ThreadPoolTaskExecutor 对象
            return builder.build();
        }
    }

    private static TaskExecutorBuilder createTskExecutorBuilder(TaskExecutionProperties properties)
    {
        // Pool 属性
        TaskExecutionProperties.Pool pool = properties.getPool();
        TaskExecutorBuilder builder = new TaskExecutorBuilder();
        builder = builder.queueCapacity(pool.getQueueCapacity());
        builder = builder.corePoolSize(pool.getCoreSize());
        builder = builder.maxPoolSize(pool.getMaxSize());
        builder = builder.allowCoreThreadTimeOut(pool.isAllowCoreThreadTimeout());
        builder = builder.keepAlive(pool.getKeepAlive());
        // Shutdown 属性
        TaskExecutionProperties.Shutdown shutdown = properties.getShutdown();
        builder = builder.awaitTermination(shutdown.isAwaitTermination());
        builder = builder.awaitTerminationPeriod(shutdown.getAwaitTerminationPeriod());
        // 其它基本属性
        builder = builder.threadNamePrefix(properties.getThreadNamePrefix());
//        builder = builder.customizers(taskExecutorCustomizers.orderedStream()::iterator);
//        builder = builder.taskDecorator(taskDecorator.getIfUnique());
        return builder;
    }

}
  • 参考 Spring Boot TaskExecutionAutoConfiguration 自动化配置类，我们创建了 ExecutorOneConfiguration 和 ExecutorTwoConfiguration 配置类，来分别创建 Bean 名字为 executor-one 和 executor-two 两个执行器。

5.4 DemoService

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
复制代码@Service
public class DemoService
{

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Async(AsyncConfig.EXECUTOR_ONE_BEAN_NAME)
    public Integer execute01()
    {
        logger.info("[execute01]");
        return 1;
    }

    @Async(AsyncConfig.EXECUTOR_TWO_BEAN_NAME)
    public Integer execute02()
    {
        logger.info("[execute02]");
        return 2;
    }

}
  • 在 @Async 注解上，我们设置了其使用的执行器的 Bean 名字。

5.5 简单测试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
复制代码@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class DemoServiceTest
{

    @Autowired
    private DemoService demoService;

    @Test
    public void testExecute() throws InterruptedException
    {
        demoService.execute01();
        demoService.execute02();

        // sleep 1 秒，保证异步调用的执行
        Thread.sleep(1000);
    }

}

运行单元测试，执行日志如下：

1
2
复制代码2020-06-08 15:38:28.846  INFO 12020 --- [     task-one-1] c.i.s.l.asynctask.service.DemoService    : [execute01]
2020-06-08 15:38:28.846  INFO 12020 --- [     task-two-1] c.i.s.l.asynctask.service.DemoService    : [execute02]
  • 从日志中，我们可以看到，#execute01() 方法在 executor-one 执行器中执行，而 #execute02() 方法在 executor-two 执行器中执行。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%