Mastering Asynchronous Processing in Java: Build a Robust SDK with Spring

Preface

A good system design must follow the open‑closed principle; as business evolves, core code changes increase error risk. Most new features extend existing functionality, requiring performance and quality, so asynchronous thread pools are often used, though they add uncertainty.

To address this, a generic asynchronous processing SDK is designed for easy implementation of various async tasks.

Purpose

Asynchronous processing ensures methods execute effectively without blocking the main flow.

More importantly, fallback mechanisms guarantee data is not lost, achieving eventual consistency.

Advantages

Non‑intrusive design with independent database, scheduled tasks, message queue, and manual execution UI (single sign‑on).

Uses Spring transaction event mechanism; async strategy failures do not affect business.

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

Even if the transaction commits and async parsing fails, fallback execution runs unless the database, message queue, or method has bugs.

Principle

After container initialization, all beans are scanned and methods annotated with @AsyncExec are cached.

At runtime, an AOP aspect publishes events for these methods.

A transaction event listener processes the asynchronous execution strategy.

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

fallbackExecution=true

– processes the event even when no transaction is active. TransactionPhase.AFTER_COMPLETION – processes the event after transaction commit or rollback.

Components

Kafka message queue

XXL‑Job scheduled tasks

MySQL database

Spring AOP

Vue UI

Design Patterns

Strategy

Template Method

Dynamic Proxy

Flowchart

Database Scripts

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

Async Strategy

Security Levels

Execution Status

Apollo Configuration

# Switch: default off
async.enabled=true

# Application name
spring.application.name=xxx

# Data source (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

# Static resources
spring.resources.add-mappings=true
spring.resources.static-locations=classpath:/static/

# Core thread count
async.executor.thread.corePoolSize=10
# Max thread count
async.executor.thread.maxPoolSize=50
# Queue capacity
async.executor.thread.queueCapacity=10000
# Keep‑alive seconds
async.executor.thread.keepAliveSeconds=600

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

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

# Retry count (default 5)
async.exec.count=5
# Retry limit
async.retry.limit=100
# Compensation limit
async.comp.limit=100
# Login interception (default false)
async.login=false

Usage

1. Async switch scm.async.enabled=true 2. Annotate methods that need async execution (must be Spring‑proxied)

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

3. Manual processing address

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

Notes

1. Application name spring.application.name 2. Queue name

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

Custom topic can be set with async.topic=xxx.

3. Business logic must be idempotent

4. One queue shared by all applications

Self‑destructing consumption.

5. Scheduled tasks

Async retry task (retry every 2 minutes, configurable retry count)

Async compensation task (compensate every hour for records older than one hour)

Demo

GitHub Repository

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