异步多线程

使用 CompletableFuture 进行异步多线程 = 👉 异步执行 + 线程池 + 回调 + 结果聚合

Demo链接

CompletableDemo: SpringBoot框架的多线程异步架构测试demo

整体架构

graph TD
    A["UserController"] --> B["UserAggregateService"]
    A --> C["UserAllAggregateService"]
    B --> D["AsyncQueryService"]
    C --> D
    D --> E["UserMapper"]
    D --> F["OrderMapper"]
    D --> G["RoleMapper"]

异步验证方法

单用户聚合接口

/user/detail?userId=xxx

UserAggregateService.aggregate() 中同时发起 3 个异步查询：

异步任务 说明
queryUser(userId) 查用户基本信息
queryOrders(userId) 查用户订单
queryRoles(userId) 查用户角色

三个任务通过 @Async("asyncExecutor") 提交到自定义线程池并行执行，然后用 CompletableFuture.allOf(...).join() 等待全部完成，最后组装成 UserAggregateDTO 返回。

案例

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
// 请求url
http://localhost:8080/user/detail?userId=3

// 返回json
{
	"user": {
		"id": 3,
		"name": "王五"
	},
	"orders": [
		{
			"id": 0,
			"userId": 3,
			"orderNo": "ORD-3-1"
		}
	],
	"roles": [
		{
			"id": 4,
			"userId": 3,
			"roleName": "GUEST"
		}
	]
}

// 控制台日志打印
queryUser thread=async-4
queryOrders userId=3 thread=async-5
queryRoles userId=3 thread=async-6
Async task finished, thread=async-6, cost=146 ms
Async task finished, thread=async-4, cost=3144 ms
Async task finished, thread=async-5, cost=3144 ms

全量用户聚合接口

/user/all-detail

UserAllAggregateService.queryAllUsersDetail() 更进一步：
先查出所有用户，对每个用户并行发起 queryOrders + queryRoles，用 thenCombine 组合结果
用 allOf().join() 等待所有用户的所有任务完成，打印总耗时。对比串行查询的耗时，并行版本的总耗时约等于最慢的那一次查询，而非所有查询耗时之和。

案例

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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
// 请求url
http://localhost:8080/user/all-detail

// 返回json
[
	{
		"userId": 1,
		"userName": "张三",
		"orders": [
			{
				"id": 0,
				"userId": 1,
				"orderNo": "ORD-1-1"
			},
			{
				"id": 0,
				"userId": 1,
				"orderNo": "ORD-1-2"
			}
		],
		"roles": [
			{
				"id": 1,
				"userId": 1,
				"roleName": "ADMIN"
			},
			{
				"id": 2,
				"userId": 1,
				"roleName": "USER"
			}
		]
	},
	{
		"userId": 2,
		"userName": "李四",
		"orders": [
			{
				"id": 0,
				"userId": 2,
				"orderNo": "ORD-2-1"
			}
		],
		"roles": [
			{
				"id": 3,
				"userId": 2,
				"roleName": "USER"
			}
		]
	},
	{
		"userId": 3,
		"userName": "王五",
		"orders": [
			{
				"id": 0,
				"userId": 3,
				"orderNo": "ORD-3-1"
			}
		],
		"roles": [
			{
				"id": 4,
				"userId": 3,
				"roleName": "GUEST"
			}
		]
	},
	{
		"userId": 4,
		"userName": "赵六",
		"orders": [
			{
				"id": 0,
				"userId": 4,
				"orderNo": "ORD-4-1"
			}
		],
		"roles": [
			{
				"id": 5,
				"userId": 4,
				"roleName": "USER"
			},
			{
				"id": 6,
				"userId": 4,
				"roleName": "DEV"
			}
		]
	}
]

// 控制台日志打印
queryOrders userId=1 thread=async-2
queryOrders userId=2 thread=async-6
queryOrders userId=3 thread=async-3
queryRoles userId=2 thread=async-8
queryRoles userId=4 thread=async-7
queryRoles userId=1 thread=async-5
queryOrders userId=4 thread=async-4
queryRoles userId=3 thread=async-1
Async task finished, thread=async-7, cost=80 ms
Async task finished, thread=async-8, cost=161 ms
Async task finished, thread=async-1, cost=161 ms
Async task finished, thread=async-5, cost=161 ms
Async task finished, thread=async-3, cost=3160 ms
Async task finished, thread=async-4, cost=3160 ms
Async task finished, thread=async-6, cost=3160 ms
Async task finished, thread=async-2, cost=6296 ms
queryAllUsersDetail cost=6435ms

注意

该示例中并没有在多线程中加入过多的try-catch异常处理和相关日志，实际在运行时可能出现异常导致线程没有释放，这样的情况要在后序实际应用时补充。