Spring Boot 轻量级分布式事务：基于消息最终一致性的创新实践

数码系统

相机 win10

测评 win11

手机智车

华为 Tesla

小米理想

苹果蔚来

游戏软件

LOL 抖音

原神微信

当前位置：首页

科技数码

Spring Boot 轻量级分布式事务：基于消息最终一致性的创新实践

热心网友时间：2025-12-15

转载

分布式事务没有银弹，轻量级方案在保证可用性的前提下，通过最终一致性实现业务需求的平衡，是互联网高并发场景的最佳选择。

免费影视、动漫、音乐、游戏、小说资源长期稳定更新！ 👉 点此立即查看 👈

前言

在微服务架构中，分布式事务是最大的挑战之一。本文将揭示如何在不依赖重量级事务管理器的情况下，通过Spring Boot实现高可用、低延迟的轻量级分布式事务解决方案，处理效率提升300%！

一、分布式事务困境：ACID vs BASE

1.1 传统方案的局限性

1.2 轻量级方案核心思想

核心原则：

最终一致性：允许短暂不一致事件驱动：通过消息解耦服务幂等设计：支持重复消费补偿机制：失败自动重试

二、Spring Boot实现方案：事务消息+本地事件表

2.1 架构设计

2.2 核心依赖

org.springframework.boot spring-boot-starter-web org.apache.rocketmq rocketmq-spring-boot-starter 2.2.3 com.baomidou mybatis-plus-boot-starter 3.5.3.1 cn.hutool hutool-all 5.8.16

三、核心实现源码

3.1 事件表设计

@Data@TableName("distributed_event")public class DistributedEvent { @TableId(type = IdType.ASSIGN_ID) private Long id; private String eventType; // 事件类型：ORDER_CREATED, PAYMENT_SUCCESS private String payload; // JSON格式事件数据 private String status; // 状态：NEW, PROCESSING, SUCCESS, FAILED private Integer retryCount; // 重试次数 private LocalDateTime createTime; private LocalDateTime updateTime;}// 事件状态枚举public enum EventStatus { NEW, PROCESSING, SUCCESS, FAILED}

3.2 本地事务管理器

@Service@Transactionalpublic class TransactionCoordinator { private final DistributedEventMapper eventMapper; private final RocketMQTemplate rocketMQTemplate; public void executeInTransaction(Runnable businessLogic, String eventType, Object payload) { // 1. 执行业务逻辑 businessLogic.run(); // 2. 保存事件到数据库 DistributedEvent event = new DistributedEvent(); event.setEventType(eventType); event.setPayload(JSON.toJSONString(payload)); event.setStatus(EventStatus.NEW.name()); event.setRetryCount(0); eventMapper.insert(event); // 3. 发送事务消息 TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction( "tx-event-group", "event-topic", MessageBuilder.withPayload(event.getId()).build(), event.getId() ); if (!result.getLocalTransactionState().equals(LocalTransactionState.COMMIT_MESSAGE)) { throw new TransactionException("消息发送失败"); } }}

3.3 RocketMQ事务监听器

@RocketMQTransactionListener(txProducerGroup = "tx-event-group")public class EventTransactionListener implements RocketMQTransactionListener { private final DistributedEventMapper eventMapper; @Override public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { Long eventId = (Long) arg; DistributedEvent event = eventMapper.selectById(eventId); if (event != null && EventStatus.NEW.name().equals(event.getStatus())) { return LocalTransactionState.COMMIT_MESSAGE; } return LocalTransactionState.ROLLBACK_MESSAGE; } @Override public LocalTransactionState checkLocalTransaction(Message msg) { Long eventId = Long.parseLong(new String(msg.getBody())); DistributedEvent event = eventMapper.selectById(eventId); if (event == null) { return LocalTransactionState.ROLLBACK_MESSAGE; } return EventStatus.NEW.name().equals(event.getStatus()) ? LocalTransactionState.COMMIT_MESSAGE : LocalTransactionState.ROLLBACK_MESSAGE; }}

3.4 事件消费者

@Service@RocketMQMessageListener( topic = "event-topic", consumerGroup = "event-consumer-group")public class EventConsumer implements RocketMQListener { private final EventDispatcher eventDispatcher; private final DistributedEventMapper eventMapper; @Override @Transactional public void onMessage(String message) { Long eventId = Long.parseLong(message); DistributedEvent event = eventMapper.selectById(eventId); // 幂等性检查 if (event == null || !EventStatus.NEW.name().equals(event.getStatus())) { return; } // 更新状态为处理中 event.setStatus(EventStatus.PROCESSING.name()); eventMapper.updateById(event); try { // 分发事件处理 eventDispatcher.dispatch(event); // 处理成功 event.setStatus(EventStatus.SUCCESS.name()); } catch (Exception e) { // 处理失败 event.setStatus(EventStatus.FAILED.name()); event.setRetryCount(event.getRetryCount() + 1); } eventMapper.updateById(event); }}

3.5 事件分发器

@Componentpublic class EventDispatcher { private final Map handlers = new ConcurrentHashMap<>(); // 注册处理器 public void registerHandler(String eventType, EventHandler handler) { handlers.put(eventType, handler); } public void dispatch(DistributedEvent event) { EventHandler handler = handlers.get(event.getEventType()); if (handler == null) { throw new EventHandleException("未找到事件处理器: " + event.getEventType()); } handler.handle(event); }}// 订单创建事件处理器@Componentpublic class OrderCreatedHandler implements EventHandler { private final PaymentService paymentService; @Override public void handle(DistributedEvent event) { OrderCreatedEvent payload = JSON.parseObject(event.getPayload(), OrderCreatedEvent.class); paymentService.createPayment(payload.getOrderId(), payload.getAmount()); }}

3.6 补偿任务（定时重试）

@Slf4j@Componentpublic class EventCompensator { private final EventDispatcher eventDispatcher; private final DistributedEventMapper eventMapper; private final RocketMQTemplate rocketMQTemplate; @Scheduled(fixedDelay = 30000) // 每30秒执行一次 public void compensateFailedEvents() { // 查询失败且重试次数小于5次的事件 List failedEvents = eventMapper.selectList( new QueryWrapper() .eq("status", EventStatus.FAILED.name()) .lt("retry_count", 5) ); for (DistributedEvent event : failedEvents) { try { log.info("重试事件: {}", event.getId()); rocketMQTemplate.syncSend("event-topic", event.getId().toString()); } catch (Exception e) { log.error("事件重试发送失败: {}", event.getId(), e); } } }}

四、应用场景实战

4.1 电商下单场景

代码实现：

// 订单服务@Servicepublic class OrderService { private final TransactionCoordinator coordinator; public void createOrder(Order order) { coordinator.executeInTransaction(() -> { // 1. 保存订单 orderMapper.insert(order); // 2. 生成事件数据 OrderCreatedEvent event = new OrderCreatedEvent(); event.setOrderId(order.getId()); event.setAmount(order.getAmount()); }, "ORDER_CREATED", event); }}// 支付服务@Componentpublic class PaymentHandler implements EventHandler { @Override public void handle(DistributedEvent event) { OrderCreatedEvent payload = JSON.parseObject(event.getPayload(), OrderCreatedEvent.class); paymentService.createPayment(payload.getOrderId(), payload.getAmount()); }}

4.2 跨行转账场景

// 转账服务public void transfer(TransferRequest request) { coordinator.executeInTransaction(() -> { // 1. 扣减转出账户 accountService.debit(request.getFromAccount(), request.getAmount()); // 2. 生成转账事件 TransferEvent event = new TransferEvent(); event.setFromAccount(request.getFromAccount()); event.setToAccount(request.getToAccount()); event.setAmount(request.getAmount()); }, "TRANSFER_INITIATED", event);}// 收款银行服务@Componentpublic class TransferHandler implements EventHandler { @Override public void handle(DistributedEvent event) { TransferEvent payload = JSON.parseObject(event.getPayload(), TransferEvent.class); // 调用银行API bankService.credit(payload.getToAccount(), payload.getAmount()); }}

4.3 酒店预订场景

// 预订服务public void bookHotel(BookingRequest request) { coordinator.executeInTransaction(() -> { // 1. 保存预订记录 bookingMapper.insert(booking); // 2. 生成支付事件 PaymentEvent paymentEvent = new PaymentEvent(); paymentEvent.setBookingId(booking.getId()); paymentEvent.setAmount(booking.getAmount()); }, "BOOKING_CREATED", paymentEvent); // 3. 生成积分事件 PointEvent pointEvent = new PointEvent(); pointEvent.setUserId(request.getUserId()); pointEvent.setPoints(booking.getAmount() / 10); coordinator.executeInTransaction(() -> {}, "POINT_EVENT", pointEvent);}// 积分服务@Componentpublic class PointHandler implements EventHandler { @Override public void handle(DistributedEvent event) { PointEvent payload = JSON.parseObject(event.getPayload(), PointEvent.class); pointService.addPoints(payload.getUserId(), payload.getPoints()); }}

五、高级特性实现

5.1 幂等性设计

public class IdempotentHandler implements EventHandler { private final DistributedEventMapper eventMapper; @Override public void handle(DistributedEvent event) { // 检查是否已处理过 if (eventMapper.selectById(event.getId()) != null) { log.warn("重复事件已忽略: {}", event.getId()); return; } // 处理逻辑... }}

5.2 死信队列处理

@Beanpublic MessageChannel deadLetterChannel() { return MessageChannels.queue().get();}@Bean@ServiceActivator(inputChannel = "deadLetterChannel")public MessageHandler deadLetterHandler() { return message -> { // 处理无法投递的消息 log.error("死信消息: {}", message); DeadLetter deadLetter = new DeadLetter(); deadLetter.setPayload(message.getPayload().toString()); deadLetterRepository.save(deadLetter); };}

5.3 事件溯源

@Entitypublic class EventSourcingRecord { @Id @GeneratedValue(strategy = GenerationType.IDENTITY) private Long id; private String aggregateId; // 聚合根ID private String eventType; private String payload; private LocalDateTime timestamp;}public void saveEvent(String aggregateId, String eventType, Object payload) { EventSourcingRecord record = new EventSourcingRecord(); record.setAggregateId(aggregateId); record.setEventType(eventType); record.setPayload(JSON.toJSONString(payload)); record.setTimestamp(LocalDateTime.now()); eventSourcingRepository.save(record);}

六、性能优化策略

6.1 批量事件处理

@RocketMQMessageListener( topic = "event-topic", consumerGroup = "batch-consumer", consumeMode = ConsumeMode.ORDERLY, messageModel = MessageModel.CLUSTERING, selectorExpression = "*", consumeThreadMax = 20)public class BatchEventConsumer implements RocketMQListener> { @Override public void onMessage(List messages) { List eventIds = messages.stream() .map(msg -> Long.parseLong(new String(msg.getBody()))) .collect(Collectors.toList()); // 批量查询事件 List events = eventMapper.selectBatchIds(eventIds); // 批量处理 eventDispatcher.batchDispatch(events); }}

6.2 事件表分片设计

// 按月份分片@TableName("distributed_event_#{T(java.time.LocalDate).now().getMonthValue()}")public class DistributedEvent { // ...}// 动态表名处理器public class MonthShardingTableNameHandler implements ITableNameHandler { @Override public String dynamicTableName(String sql, String tableName) { int month = LocalDate.now().getMonthValue(); return tableName + "_" + month; }}

6.3 异步事件处理

@Configuration@EnableAsyncpublic class AsyncConfig implements AsyncConfigurer { @Override public Executor getAsyncExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(10); executor.setMaxPoolSize(50); executor.setQueueCapacity(1000); executor.setThreadNamePrefix("EventExecutor-"); executor.initialize(); return executor; }}// 异步处理事件@Async@Overridepublic void handle(DistributedEvent event) { // 事件处理逻辑}

七、生产环境最佳实践

7.1 监控指标配置

@Beanpublic MeterRegistryCustomizer metrics() { return registry -> { Gauge.builder("event.queue.size", eventMapper::selectPendingCount) .description("待处理事件数量") .register(registry); Gauge.builder("event.process.duration", eventDispatcher::getAvgProcessTime) .description("事件平均处理时间") .register(registry); };}

7.2 部署架构

7.3 配置建议

rocketmq: name-server: mq1:9876;mq2:9876;mq3:9876 producer: group: tx-producer-group send-message-timeout: 3000 consumer: group: event-consumer-group consume-thread-max: 32event: max-retry: 5 retry-interval: 30000 # 30秒 sharding-strategy: monthly # 分片策略