How to Build a Robust Asynchronous Processing SDK with Spring, Kafka, and XXL‑Job

Preface

Good system design follows the open‑closed principle. As business evolves, core code changes frequently, increasing error risk. Most new features extend existing functionality and use asynchronous thread pools, which introduce uncertainty. To address this, a generic asynchronous processing SDK was designed to simplify various async scenarios.

Purpose

The SDK ensures that asynchronous methods are reliably executed without blocking the main flow, while providing fallback mechanisms to prevent data loss and achieve eventual consistency.

Advantages

Non‑intrusive design.

Independent database, scheduled tasks, message queue, and manual execution UI (single sign‑on).

Leverages Spring transaction event mechanism so that async strategy failures do not affect business logic.

If a method runs within a transaction, the event is processed after transaction commit or rollback.

Even after transaction commit, failures in async strategy are handled by fallback mechanisms (unless DB, MQ, or the method itself fails).

Principle

After Spring finishes bean initialization, it scans all methods for the @AsyncExec annotation and caches them. When a cached method is invoked, an AOP aspect publishes an event. A @TransactionalEventListener with fallbackExecution=true and phase=TransactionPhase.AFTER_COMPLETION processes the async execution, handling both after commit and after rollback.

Components

Kafka message queue

XXL‑Job scheduled tasks

MySQL database

Spring AOP

Vue front‑end UI

Design Patterns

Strategy pattern

Template method

Dynamic proxy

Async Strategy Configuration (Apollo)

# Switch (default off)
async.enabled=true

# Application name
spring.application.name=xxx

# DataSource (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

# Thread pool settings
async.executor.thread.corePoolSize=10
async.executor.thread.maxPoolSize=50
async.executor.thread.queueCapacity=10000
async.executor.thread.keepAliveSeconds=600

# Delete record after successful execution (default true)
async.exec.deleted=true

# Topic prefix (default application name)
async.topic=${spring.application.name}_async_queue

# Retry settings
async.exec.count=5
async.retry.limit=100
async.comp.limit=100

# Login interception (default false)
async.login=false

Usage

Enable async switch: scm.async.enabled=true Add

@AsyncExec(type=AsyncExecEnum.SAVE_ASYNC, remark="Data Dictionary")

to any Spring‑proxied method that needs asynchronous execution.

Access the manual processing UI at http://localhost:8004/async/index.html.

Notes

Set spring.application.name correctly; it is used to build the queue name.

Queue name pattern: ${async.topic:${spring.application.name}}_async_queue Ensure business logic is idempotent.

One queue per application is recommended (self‑produce, self‑consume).

Two scheduled jobs handle retries and compensation:

Async retry job runs every 2 minutes with configurable retry count.

Async compensation job runs hourly for records older than one hour.

Database Schema

CREATE TABLE `async_scene` (
  `id` bigint NOT NULL AUTO_INCREMENT COMMENT '主键ID',
  `application_name` varchar(100) NOT NULL DEFAULT '' COMMENT '应用名称',
  `method_sign` varchar(50) NOT NULL DEFAULT '' COMMENT '方法签名',
  `scene_name` varchar(200) NOT NULL DEFAULT '' COMMENT '业务场景描述',
  `async_type` varchar(50) NOT NULL DEFAULT '' COMMENT '异步策略类型',
  `queue_name` varchar(200) NOT NULL DEFAULT '' COMMENT '队列名称',
  `theme_value` varchar(100) NOT NULL DEFAULT '' COMMENT '消费主题',
  `exec_count` int NOT NULL DEFAULT '0' COMMENT '失败重试次数',
  `exec_deleted` int NOT NULL DEFAULT '0' COMMENT '执行后是否删除',
  `async_version` varchar(50) NOT NULL DEFAULT '' COMMENT '组件版本号',
  `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '更新时间',
  `cdc_crt_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '记录新增时间',
  `cdc_upd_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '记录修改时间',
  PRIMARY KEY (`id`) USING BTREE,
  UNIQUE KEY `uk_application_sign` (`application_name`,`method_sign`) USING BTREE,
  KEY `idx_cdc_upd_time` (`cdc_upd_time`)
) ENGINE=InnoDB COMMENT='异步场景表';

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='异步处理日志';

Effect Demonstration

UI and processing‑flow screenshots:

Code Repository

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