Appearance

异步与定时任务 ​

ECShopX Java 有两套互补的后台机制：

机制模块典型场景
Dispatch 异步总线ecshopx-dispatch业务触发的 Job、跨模块异步事件消费
XXL-JOB 定时任务各 Bundle 的 cron 包按 Cron 扫表、对账、状态轮询

RabbitMQ：DispatchDriverType 枚举含 RABBITMQ，但当前生产装配以 Sync + Redis 为主（见 DispatchBusConfig）；Rabbit 驱动存在但未作为默认路径，二次开发无需优先接入。

A. ecshopx-dispatch ​

架构概览 ​

业务 Service / Publisher
        │
        ▼
  DispatchFacade          ← 统一入口（publishEvent / dispatchJob）
        │
        ▼
   DispatchCore           ← 按 DispatchMode 路由
    ├─ SYNC  → SyncDispatchDriver（当前线程执行）
    └─ ASYNC → RedisDispatchDriver（入队）
                    │
                    ▼
         DispatchConsumerRuntime（出队消费、重试、失败记录）
                    │
                    ▼
         DispatchRegistry → DispatchListener / DispatchHandler

Spring 装配入口：ecshopx-bootstrap/.../config/DispatchBusConfig.java。

入口：DispatchFacade ​

方法消息类型用途
publishEvent(messageName, payload, options)EVENT领域事件 fan-out
dispatchJob(messageName, payload, options)JOB后台任务（导出、同步分页等）

二者均构造 DispatchMessage，经 DispatchCore#dispatch 路由。

驱动类型：DispatchDriverType ​

值行为
SYNC同步执行；SyncDispatchDriver 直接查 DispatchRegistry 并调用 handler / listener
REDIS异步入队；RedisDispatchDriver 写入 Redis List（就绪队列）或 ZSet（延迟队列）
RABBITMQ预留；非当前主路径

DispatchOptions 常用工厂：

java
DispatchOptions.eventDefaults();      // 同步事件
DispatchOptions.jobDefaults();        // 异步 Job，默认重试
DispatchOptions.oplQueuedAfterCommit(); // 订单流程日志：异步 + Redis

发布 Job（异步任务） ​

Job 与 Event 共用总线，区别在于 messageType == JOB 且注册的是 DispatchHandler。

注册 handler（bootstrap）：

java
// DrawCashJobRegistrationConfig
dispatchRegistry.registerJob(AdaPayDispatchJobNames.DRAW_CASH_JOB, drawCashJobHandler);

实现 handler（业务模块 dispatch 包）：

java
@Component
public class GetItemsSpecFromOmeJobHandler implements DispatchHandler {
    @Override
    public void handle(Map<String, Object> payload) {
        pagedSyncRunner.consumeQueuedPage(payload);
    }
}

投递 Job（bootstrap Publisher 或 Service）：

java
dispatchFacade.dispatchJob(jobName, payload, DispatchOptions.jobDefaults());

延迟投递 ​

异步消息可在 DispatchOptions 或 ListenerDispatchOptions 中设置 Duration delay：

  • RedisDispatchDriver#enqueue：若配置了 dispatch.redis.delayed-zset-key（默认 dispatch:delayed）且 delay > 0，消息写入 ZSet，score 为到期时间戳。
  • RedisDispatchDelayedPollScheduler 定时调用 RedisDelayedDispatchMover#moveReadyMessages，将到期消息移入就绪队列 dispatch:{queue}。

重试 ​

组件职责
RetryPolicymaxAttempts（默认 3）+ nextDelay（默认 3s）；RetryPolicy.platformDefault()
DispatchRetryDecidercurrentAttempt < maxAttempts 则重试
RedisDispatchDriver#retry重新入队
DispatchConsumerRuntime#consume捕获异常 → 重试或失败

重试次数耗尽后：

  1. FailedJobRecorder#recordFailure 写入失败任务表（FailedJobEntity / FailedJobMapper）；默认 bootstrap 中 DispatchBusConfig#failedJobMapper 为 no-op 桩实现（return 1），生产若需落库须替换为真实 Mapper 实现
  2. RedisDispatchDriver#fail 将消息推入 死信队列 dispatch:dead:{queue}

消费运行时 ​

DispatchConsumerRuntime 根据 messageType 分发：

  • JOB → registry.jobHandler(messageName).handle(payload)
  • EVENT → 按 listenerName 精确匹配单个 DispatchListener#onEvent

成功：stateRecorder.recordAck；失败且不重试：failedJobRecorder + stateRecorder.recordFail。

二次开发：加异步 Job 步骤 ​

  1. 在 ecshopx-common/.../*DispatchJobNames 定义 job 名（或与已有常量对齐的字符串）。
  2. 业务模块 dispatch 包实现 DispatchHandler。
  3. ecshopx-bootstrap/config/*JobRegistrationConfig 中 registerJob。
  4. 需要投递处调用 dispatchFacade.dispatchJob(...)，按需指定 queue、delay、RetryPolicy。
  5. 确保 Redis 可用（dispatchBusStringRedisTemplate）；本地见 快速入门 Compose 或 local profile。

二次开发：加异步事件监听 ​

见 事件系统 注册 DispatchListener 一节；异步 listener 的队列名在 ListenerDispatchOptions.async(queue, delay) 中指定。

B. XXL-JOB 定时任务 ​

角色分工 ​

组件部署职责
XXL-JOB Admin独立进程 / Compose 服务Cron 调度、任务配置、执行日志
Executor随 ecshopx-java 启动注册到 Admin，接收触发并执行 @XxlJob 方法

Admin 地址（注意 Compose vs 本地 profile） ​

环境Admin URL
Docker Compose（docker-compose.yml）http://localhost:8080/xxl-job-admin
非 Compose / local profile（application.properties）http://127.0.0.1:9080/xxl-job-admin

Compose 内应用通过 --xxl.job.admin.addresses=http://xxl-job-admin:8080/xxl-job-admin 连接 Admin；与宿主机浏览器访问的 8080 映射一致。

Executor 端口 ​

应用侧执行器默认端口 19999（application.properties 中 xxl.job.executor.port=19999）。Admin 调度时需能访问该端口（Compose 网络内通常自动可达）。

关键配置（ecshopx-bootstrap/src/main/resources/application.properties）：

properties
xxl.job.admin.addresses=http://127.0.0.1:9080/xxl-job-admin
xxl.job.accessToken=ecshopx-cron-dev
xxl.job.executor.appname=ecshopx-executor
xxl.job.executor.port=19999

accessToken 须与 Admin 侧一致（Compose 中 xxl-job-admin 服务同样使用 ecshopx-cron-dev）。

编写 Handler ​

在业务模块 cron 包声明 Spring @Component，方法标注 @XxlJob("handler-name")。handler 名必须与 Admin 中 JobHandler 完全一致。

真实示例（ecshopx-aliyunsms）：

java
@Slf4j
@Component
@RequiredArgsConstructor
public class ScheduleRunTaskHandler {

    private final TaskService taskService;
    private final ApplicationEventPublisher eventPublisher;

    @XxlJob("aliyunsms-run-task")
    public void execute() {
        long start = System.currentTimeMillis();
        try {
            int n = taskService.scheduleRunTask();
            log.info("[cron][aliyunsms-run-task] done, cost={}ms, taskRowsIsSendSet={}",
                    System.currentTimeMillis() - start, n);
        } catch (Exception e) {
            log.error("[cron][aliyunsms-run-task] failed, cost={}ms",
                    System.currentTimeMillis() - start, e);
            eventPublisher.publishEvent(new CronAlertEvent("aliyunsms-run-task", e));
            throw e;
        }
    }
}

订单域另有 @XxlJob("finish-orders") 等 handler，模式相同：委托 *CronService，记录耗时，失败时发布 CronAlertEvent 并 rethrow 以便 Admin 标记失败。

Admin 配置步骤（联调） ​

  1. 启动 MySQL / Redis / XXL-JOB Admin / ecshopx-java（推荐 docker compose up -d）。
  2. 浏览器打开 Admin（Compose：http://localhost:8080/xxl-job-admin）。
  3. 执行器管理：确认 ecshopx-executor 已注册（应用日志无 token 错误）。
  4. 任务管理：新增任务，JobHandler 填 @XxlJob 注解值（如 aliyunsms-run-task），配置 Cron，路由策略选「第一个」或指定执行器。
  5. 手动触发一次，在 Admin「调度日志」与业务日志中验证。

Dispatch Job vs XXL-JOB：如何选型 ​

问题选 Dispatch Job选 XXL-JOB
由用户操作 / API 触发？✅❌
固定 Cron 扫表？❌✅
需要 Admin 可视化 Cron / 手工补跑？❌✅
需要与 dispatch 事件总线共用 Redis 队列？✅❌

相关章节 ​