萬能的異步處理方案

作者：三火哥

原文：https://juejin.cn/post/7266087843239084090

1 前言

良好的系統設計必須要做到開閉原則，隨着業務的不斷迭代更新，核心代碼也會被不斷改動，出錯的概率也會大大增加。但是大部分增加的功能都是在擴展原有的功能，既要保證性能又要保證質量，我們往往都會使用異步線程池來處理，然而卻增加了很多不確定性因素。

由此我設計了一套通用的異步處理 SDK，可以很輕鬆的實現各種異步處理

2 目的

通過異步處理不僅能夠保證方法能夠得到有效的執行而且不影響主流程

更重要的是各種兜底方法保證數據不丟失，從而達到最終一致性

3 優點

無侵入設計，獨立數據庫，獨立定時任務，獨立消息隊列，獨立人工執行界面 (統一登錄認證)

使用 spring 事務事件機制，即使異步策略解析失敗也不會影響業務

如果你的方法正在運行事務，會等事務提交後或回滾後再處理事件

就算事務提交了，異步策略解析失敗了，我們還有兜底方案執行（除非數據庫有問題，消息隊列有問題，方法有 bug）

4 原理

容器初始化 bean 完成後遍歷所有方法，把有 @AsyncExec 註解的方法緩存起來

方法運行時通過 AOP 切面發佈事件

事務事件監聽處理異步執行策略

@TransactionalEventListener(fallbackExecution = true, phase = TransactionPhase.AFTER_COMPLETION)

fallbackExecution=true 沒有事務正在運行，依然處理事件
TransactionPhase.AFTER_COMPLETION 事務提交後和事務回滾後都處理事件

5 組件

kafka 消息隊列
xxl job 定時任務
mysql 數據庫
spring 切面
vue 界面

6 設計模式

策略
模板方法
動態代理

7 流程圖

8 數據庫腳本

CREATE TABLE `async_req` (
  `id` bigint NOT NULL AUTO_INCREMENT COMMENT '主鍵ID',
  `application_name` varchar(100) NOT NULL DEFAULT '' COMMENT '應用名稱',
  `sign` varchar(50) NOT NULL DEFAULT '' COMMENT '方法簽名',
  `class_name` varchar(200) NOT NULL DEFAULT '' COMMENT '全路徑類名稱',
  `method_name` varchar(100) NOT NULL DEFAULT '' COMMENT '方法名稱',
  `async_type` varchar(50) NOT NULL DEFAULT '' COMMENT '異步策略類型',
  `exec_status` tinyint NOT NULL DEFAULT '0' COMMENT '執行狀態 0：初始化 1：執行失敗 2：執行成功',
  `exec_count` int NOT NULL DEFAULT '0' COMMENT '執行次數',
  `param_json` longtext COMMENT '請求參數',
  `remark` varchar(200) NOT NULL DEFAULT '' COMMENT '業務描述',
  `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '創建時間',
  `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '更新時間',
  PRIMARY KEY (`id`) USING BTREE,
  KEY `idx_applocation_name` (`application_name`) USING BTREE,
  KEY `idx_exec_status` (`exec_status`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='異步處理請求';

CREATE TABLE `async_log` (
  `id` bigint NOT NULL AUTO_INCREMENT COMMENT '主鍵ID',
  `async_id` bigint NOT NULL DEFAULT '0' COMMENT '異步請求ID',
  `error_data` longtext COMMENT '執行錯誤信息',
  `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '創建時間',
  PRIMARY KEY (`id`) USING BTREE,
  KEY `idx_async_id` (`async_id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='異步處理日誌';

9 異步策略

10 安全級別

11 執行狀態

12 流程圖

13apollo 配置

# 開關：默認關閉
async.enabled=true

# 應用名稱
spring.application.name=xxx

# 數據源 druid 
spring.datasource.driver-class-name=com.mysql.jdbc.Driver
spring.datasource.url=jdbc:mysql://127.0.0.1:3306/fc_async?useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull&useSSL=false&allowMultiQueries=true&rewriteBatchedStatements=true
spring.datasource.username=user
spring.datasource.password=xxxx
spring.datasource.filters=config
spring.datasource.connectionProperties=config.decrypt=true;config.decrypt.key=yyy
#靜態地址
spring.resources.add-mappings=true
spring.resources.static-locations=classpath:/static/


# 以下配置都有默認值
# 核心線程數
async.executor.thread.corePoolSize=10
# 最大線程數
async.executor.thread.maxPoolSize=50
# 隊列容量
async.executor.thread.queueCapacity=10000
# 活躍時間
async.executor.thread.keepAliveSeconds=600

# 執行成功是否刪除記錄：默認刪除
async.exec.deleted=true
 
# 自定義隊列名稱前綴：默認應用名稱
async.topic=${spring.application.name}
 
# 重試執行次數：默認5次
async.exec.count=5
 
# 重試最大查詢數量
async.retry.limit=100

# 補償最大查詢數量
async.comp.limit=100

# 登錄攔截：默認false
async.login=false

14 用法

1，異步開關

scm.async.enabled=true

2，在需要異步執行的方法加註解（必須是 spring 代理方法）

@AsyncExec(type = AsyncExecEnum.SAVE_ASYNC, remark = "數據字典")

3，人工處理地址

http://localhost:8004/async/index.html

15 注意

1，應用名稱

spring.application.name

2，隊列名稱

${async.topic:${spring.application.name}}_async_queue

自定義 topic：async.topic=xxx

3，自己業務要做冪等

4，一個應用公用一個隊列

自產自消

5，定時任務

異步重試定時任務（2 分鐘重試一次，可配置重試次數）
異步補償定時任務（一小時補償一次，創建時間在一小時之前的）

16 效果展示

17github 地址

https://github.com/xiongyanokok/fc-async.git

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/moJxN3UeIwve9fxsp9wcVQ