How to Build a Robust Asynchronous Processing SDK with Spring and Kafka

Introduction

A good system design must follow the open‑closed principle. As business evolves, core code changes frequently, increasing the risk of errors. Most new features extend existing functionality, requiring both performance and quality, so asynchronous thread pools are often used, which adds uncertainty.

To address this, a generic asynchronous processing SDK was designed to simplify various async tasks.

Purpose

Asynchronous processing ensures that methods execute effectively without blocking the main flow, while guaranteeing data consistency through fallback mechanisms.

Advantages

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

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

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

Even after transaction commit, failures are handled by fallback plans unless the database, queue, or method itself fails.

Principle

After container initialization, all beans are scanned; methods annotated with @AsyncExec are cached. At runtime, an AOP aspect publishes an event, and a transactional 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 handles the event after transaction commit or rollback.

Components

Kafka message queue

XXL‑Job scheduler

MySQL database

Spring AOP aspect

Vue UI

Design Patterns

Strategy

Template Method

Dynamic Proxy

Flowchart

Database Scripts

CREATE TABLE `async_scene` ( ... ) ENGINE=InnoDB COMMENT='异步场景表';
CREATE TABLE `async_req` ( ... ) ENGINE=InnoDB COMMENT='异步处理请求';
CREATE TABLE `async_log` ( ... ) ENGINE=InnoDB COMMENT='异步处理日志';

Asynchronous Strategy

Security Level

Execution Status

Flowcharts

Apollo Configuration

# 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

# 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
# Topic name (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 intercept (default false)
async.login=false

Usage

Enable async: scm.async.enabled=true Add @AsyncExec to Spring‑proxied methods, e.g.

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

Manual handling URL:

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

Notes

Set spring.application.name correctly.

Define a unique queue name, e.g. ${async.topic:${spring.application.name}}_async_queue.

Ensure business logic is idempotent.

One queue per application.

Self‑consumption pattern.

Scheduled tasks: retry every 2 minutes (configurable count) and compensation every hour for records older than one hour.

Effect Demonstration

Code Repository

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