在 Spring Event 接收到事件消息后，无论是 MQ 的发送，还是 HTTP 的调用，都是有可能失败的。可能是网络超时，可能是服务宕机，可能是线程阻塞，可能是流量洪峰等等原因。
所以，我们需要对完成 MQ、HTTP 的处理后，首先更新数据库任务表状态，是成功还是失败。之后再由定时扫描任务来做补偿处理，以确保最终一致性。
- 首先，任务补偿，它是一个定时任务扫描库表的过程。这里为了提高整体的扫描效率，设计了门牌号，可以配置多个任务，每个任务扫描自己的门牌号范围。另外扫描库表，还要根据条件获取一个最小的符合的id，之后在 > id limit x 获取数据列表。
- 然后，还要对之前的流程做一些补充。http 调用操作、MQ 推送操作，之后要做数据库表任务记录的更新，成功或者失败。这里看，从 Spring Event 接收到消息后，执行通知操作（http、mq），之后更新数据库。这些都不是一个事务的，也就是说从事务（业务数据+任务表数据写入）往后，都是有可能失败的。所以要做任务补偿操作，那么这里也带来一个问题，补偿就有可能重复，比如 http 重复调用一次，mq 重复发送一次，所以这些业务一定要做幂等操作（比如 OrderId 做唯一索引处理）。
编码实现
- 以 TaskMessageEventJob 为入口，基于 LocalTaskMessageAutoProperties 任务配置拆分任务扫描数据门牌号范围。之后调用新增加的 ILocalTaskMessageDataService 服务，处理数据的扫描拿到待处理的任务数据，发送 SpringEvent 事件。事件发送后，继续走之前实现的监听流程。
在 HttpNotifyStrategy、RabbitmqMqNotifyStrategy，两个流程处理中，补充添加更新数据库任务表操作，成功或失败。
数据操作（Dao）
public interface ITaskMessageDao { /** * 插入任务消息 * * @param taskmessagePO 任务消息PO对象 * @return 影响行数 * @throws SQLException */ int insert(TaskMessagePO taskmessagePO) throws SQLException; /** * 根据任务ID修改状态 * * @param taskId 任务ID * @param status 状态（0-待处理，1-处理中，2-已完成，3-失败） * @return 影响行数 */ int updateStatusByTaskId(String taskId, Integer status); /** * 根据门牌号查询任务消息列表 * * @param houseNumbers 门牌号列表 * @param id 查询ID大于此值的记录 * @param limit 限制返回结果数量 * @return 任务消息列表 */ List<TaskMessagePO> selectByHouseNumber(List<Integer> houseNumbers, Long id, Integer limit); /** * 根据门牌号查询符合条件的最小ID * * @param houseNumbers 门牌号列表 * @return 最小ID，如果没有找到则返回NULL */ Long selectMinIdByHouseNumber(List<Integer> houseNumbers); }任务更新
@Slf4j @Component("httpNotifyStrategy") public class HttpNotifyStrategy implements INotifyStrategy { @Resource private ILocalTaskMessagePort port; @Resource private ILocalTaskMessageRepository repository; @Override public String notify(TaskMessageEntityCommand command) throws Exception { try { String result = port.notify2http(command); repository.updateTaskStatusToSuccess(command.getTaskId()); return result; } catch (Exception e) { log.error("http notify error {}", JSON.toJSONString(command), e); // 通知失败，更新状态为失败 repository.updateTaskStatusToFailed(command.getTaskId()); throw e; } } } @Slf4j @Component("rabbitMqNotifyStrategy") public class RabbitMqNotifyStrategy implements INotifyStrategy { @Resource private ILocalTaskMessagePort port; @Resource private ILocalTaskMessageRepository repository; @Override public String notify(TaskMessageEntityCommand command) throws Exception { try { String result = port.notify2rabbitmq(command); repository.updateTaskStatusToSuccess(command.getTaskId()); return result; } catch (Exception e) { log.error("rabbitmq notify error {}", JSON.toJSONString(command), e); // 通知失败，更新状态为失败 repository.updateTaskStatusToFailed(command.getTaskId()); throw e; } } }扫描任务
因为不同的任务可以扫描不同的门牌号，这里要增加一个自定义属性配置，允许用户根据自己的诉求，拆分出多个任务。
@Data @ConfigurationProperties(prefix = "local.task.message.config", ignoreInvalidFields = true) public class LocalTaskMessageAutoProperties { /** * 任务组配置列表 */ private List<TaskGroupConfig> groups = new ArrayList<>(); @Data public static class TaskGroupConfig { /** * 任务组ID或名称，用于日志区分 */ private String groupId = "default"; /** * 扫描的门牌号列表，例如 [1,2,3] */ private List<Integer> houseNumbers = new ArrayList<>(); /** * 调度 cron 表达式，例如："0/10 * * * * ?" 表示每10秒 */ private String cron; /** * 固定延迟毫秒；若配置此项则使用固定延迟调度 */ private Long fixedDelayMs; /** * 每次批量处理限制条数 */ private Integer limit = 100; } }配置实例：
local: task: message: config: groups: # 使用 cron 表达式调度（每 10 秒执行一次） - groupId: group01 houseNumbers: [1, 2, 3] cron: "0/10 * * * * ?" limit: 100 # 使用固定延迟调度（每 5 秒执行一次） - groupId: group02 houseNumbers: [4, 5] fixedDelayMs: 5000 limit: 50触发层新增
TaskMessageEventJob扫描任务@Slf4j @Component @RequiredArgsConstructor public class TaskMessageEventJob { private final Map<String, AtomicLong> groupLastIdMap = new ConcurrentHashMap<>(); private final LocalTaskMessageAutoProperties properties; private final ThreadPoolTaskScheduler scheduler; private final ILocalTaskMessageNotifyService notifyService; private final ILocalTaskMessageDataService dataService; @PostConstruct public void init() { List<LocalTaskMessageAutoProperties.TaskGroupConfig> groups = properties.getGroups(); if (groups == null || groups.isEmpty()) { log.info("TaskMessageEventJob 未配置任务组，跳过调度初始化"); return; } for (LocalTaskMessageAutoProperties.TaskGroupConfig group : groups) { scheduleGroup(group); } } private void scheduleGroup(LocalTaskMessageAutoProperties.TaskGroupConfig group) { String groupId = group.getGroupId(); List<Integer> houseNumbers = group.getHouseNumbers(); if (houseNumbers == null || houseNumbers.isEmpty()) { log.warn("任务组 [{}] 未配置 houseNumbers,跳过该组调度", groupId); return; } // 初始化lastId groupLastIdMap.computeIfAbsent(groupId, k -> { Long minId = dataService.selectMinIdByHouseNumber(houseNumbers); long startId = (minId == null ? 0L : minId); log.info("任务组 [{}] 初始化起始ID为 {}，houseNumbers={}", groupId, startId, houseNumbers); return new AtomicLong(startId); }); Runnable task = () -> { try { long lastId = groupLastIdMap.get(groupId).get(); List<TaskMessageEntityCommand> list = dataService.selectByHouseNumber(houseNumbers, lastId, group.getLimit()); if (list == null || list.isEmpty()) { return; } for (TaskMessageEntityCommand cmd : list) { notifyService.notify(cmd); } Long maxId = list.stream().map(TaskMessageEntityCommand::getId).max(Comparator.naturalOrder()).orElse(lastId); groupLastIdMap.get(groupId).set(maxId); log.info("任务组 [{}] 处理完成，拉取 {} 条，lastId:{} -> {}", groupId, list.size(), lastId, maxId); } catch (Exception e) { log.error("任务组 [{}] 执行异常：{}", groupId, e.getMessage(), e); } }; if (group.getCron() != null && !group.getCron().trim().isEmpty()) { scheduler.schedule(task, new CronTrigger(group.getCron())); log.info("任务组 [{}] 已按 cron [{}] 调度", groupId, group.getCron()); } else { long delay = group.getFixedDelayMs() != null ? group.getFixedDelayMs() : 5000L; scheduler.scheduleWithFixedDelay(task, delay); log.info("任务组 [{}] 已按 fixedDelayMs [{}] 调度", groupId, delay); } } }- 类职责：TaskMessageEventJob 是一个“定时拉取并发布任务消息”的调度器。按照配置的任务组定期查询待处理消息，发布事件，并记录每个任务组已处理到的最大 id，用于增量继续拉取，实现动态补偿。
核心功能：
- groupLastIdMap：保存“任务组 → 最近处理的最大ID”的映射，用于增量拉取，避免重复处理。类型为 AtomicLong，保证并发场景下读写的原子性。
- properties：读取外部配置（任务组、门牌号、调度方式等）。
- scheduler：Spring 的线程池调度器，负责按 cron 或固定延迟执行任务。
- notifyService：对外发布事件（通知消费方处理每条消息）。
- dataService：数据查询服务，从存储中按条件拉取消息记录。
初始化流程：
- 读取配置中的 groups；如果未配置，打印日志并跳过初始化。
- 对每个任务组调用 scheduleGroup(group) 进行独立的调度注册。
注册任务组调度：
- 参数校验：必须配置 houseNumbers（门牌号集合），否则跳过。
初始化增量起点 `lastId：
- 使用 dataService.selectMinIdByHouseNumber(houseNumbers) 找到该组关注的门牌号里的最小 id。
- 若查询为空则从 0 开始；将起始值保存到 groupLastIdMap，供后续增量拉取使用。
构造执行体 task：每次调度触发时，按以下步骤处理：
- 读取当前组的 lastId。
- 通过 dataService.selectByHouseNumber(houseNumbers, lastId, group.getLimit()) 拉取“id 大于 lastId”的消息，最多 limit 条。
- 如果本次没有新消息，直接返回。
- 遍历消息并发布事件：notifyService.notify(cmd)。
- 计算本批次的最大 id 并更新 groupLastIdMap，作为下次拉取的起点。
- 打印处理日志，异常捕获与记录。
调度方式选择：
- 如果配置了 cron 表达式，则按 CronTrigger 定时执行。
- 否则按固定延迟 fixedDelayMs 执行；未配置时默认 5000ms。
增量与补偿的意义：
- 通过记录 lastId，每次只拉取“新产生”的消息，实现增量处理，避免重复。
- 如果某次运行异常，中断前更新的 lastId 仍在内存中；下次调度会从该值继续，尽量减少漏处理。但为更强的容灾，实际工程中常把 lastId 持久化（例如数据库/Redis），并配合消费方的幂等设计，确保“最多一次”或“至少一次”语义。
配置示例说明：
- group01 使用 cron（每10秒执行），拉取门牌号 [1,2,3]，每次最多 limit=100 条。
- group02 使用固定延迟（每5秒执行），拉取门牌号 [4,5]，每次最多 limit=50 条。
