Universal Async Processing SDK: Spring Transaction Events + Kafka + XXL-Job for Guaranteed Execution

Background and Purpose

As business iterates, core code changes frequently, increasing error risk. New features often extend existing functionality, requiring both performance and quality. Async thread pools are commonly used but introduce uncertainty. The author designed a universal async processing SDK (fc-async) to handle various async scenarios easily.

The goal: ensure methods execute effectively without blocking the main flow, and provide multiple fallback mechanisms to guarantee no data loss, achieving eventual consistency.

Key Advantages

Non-invasive design: independent database, independent scheduled tasks (XXL-Job), independent message queue (Kafka), independent manual execution UI (Vue, unified auth).

Uses Spring transaction event mechanism: even if async strategy parsing fails, the business transaction is unaffected.

If a method runs within a transaction, the event is processed after commit or rollback ( TransactionPhase.AFTER_COMPLETION). fallbackExecution=true ensures events are handled even when no transaction is running.

Multiple fallback layers: if transaction commits but async parsing fails, fallback execution still runs (unless DB, MQ, or method itself has bugs).

Core Principle

After container initializes beans, scan all methods and cache those annotated with @AsyncExec.

At runtime, an AOP aspect publishes an event when the annotated method is invoked.

A @TransactionalEventListener with fallbackExecution=true and phase=TransactionPhase.AFTER_COMPLETION listens and processes the async execution strategy.

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

fallbackExecution=true

means the listener fires even without an active transaction. AFTER_COMPLETION fires after both commit and rollback.

Components

Kafka message queue

XXL-Job scheduled tasks

MySQL database

Spring AOP

Vue front-end UI

Design Patterns Used

Strategy pattern

Template Method pattern

Dynamic Proxy

Flowchart-driven design

Database Schema

Three tables are created:

async_scene (Async Scene Configuration)

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='异步场景表';

async_req (Async Request Records)

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='异步处理请求';

async_log (Async Execution Logs)

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

Async Strategies, Security Levels, Execution States, Flowcharts

The article includes several diagrams (images) illustrating: async strategy types, security level classification, execution state transitions, and end-to-end flowcharts. These are not reproduced in text but are referenced as core design artifacts.

Apollo Configuration

All configuration properties have defaults; key ones include:

# 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 # Thread pool async.executor.thread.corePoolSize=10 async.executor.thread.maxPoolSize=50 async.executor.thread.queueCapacity=10000 async.executor.thread.keepAliveSeconds=600 # Delete on success (default true) async.exec.deleted=true # Custom queue prefix (default ${spring.application.name}) async.topic=${spring.application.name} # Retry count (default 5) async.exec.count=5 # Retry max query limit async.retry.limit=100 # Compensation max query limit async.comp.limit=100 # Login intercept (default false) async.login=false

Usage

Enable async: scm.async.enabled=true (or async.enabled=true).

Annotate the target Spring-proxied method:

@AsyncExec(type = AsyncExecEnum.SAVE_ASYNC, remark = "数据字典")

Manual handling UI: http://localhost:8004/async/index.html (requires login if async.login=true).

Important Notes

Application name ( spring.application.name) must be set uniquely per service.

Queue name format: ${async.topic:${spring.application.name}}_async_queue — one queue per application (producer and consumer are the same app).

Business logic must be idempotent because retries can cause duplicate execution.

Two scheduled jobs:

Async retry job — runs every 2 minutes, retries failed tasks up to configured count ( async.exec.count).

Async compensation job — runs hourly, picks up requests created more than 1 hour ago that are still not successful, up to async.comp.limit records.

Effect Screenshots

The article shows two UI screenshots (images) demonstrating the manual execution interface and request list with status.

Source Code

GitHub repository:

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