基于本地消息表实现分布式事务

目前做的这个项目中，不能引入其他的外部组件，故用本地消息表来实现分布式事务。下面是一个示例：
假设我们有一个电商系统，包含订单服务和库存服务。当用户下单时，需要在订单服务中创建订单，同时在库存服务中扣减库存。这是一个典型的分布式事务场景，我们需要保证这两个操作要么都成功，要么都失败，以保证数据的最终一致性。

项目结构

订单服务(Order Service)
库存服务(Inventory Service)
本地消息表(Local Message Table)
消息恢复系统(Message Recovery System)

核心思想

使用本地消息表来实现分布式事务。在订单服务中，我们将创建订单和发送消息这两个操作放在一个本地事务中。如果本地事务成功，则订单创建成功，消息也被保存到本地消息表中。然后通过定时任务或消息队列来发送消息到库存服务，实现库存扣减。如果在这个过程中出现任何异常，我们可以通过重试机制来保证最终一致性。

详细代码实现

订单服务(Order Service)

@Service
@Transactional
public class OrderService {

    @Autowired
    private OrderRepository orderRepository;

    @Autowired
    private LocalMessageRepository localMessageRepository;

    @Autowired
    private KafkaTemplate kafkaTemplate;

    public void createOrder(Order order) {
        // 开启本地事务
        TransactionStatus txStatus = transactionManager.getTransaction(new DefaultTransactionDefinition());

        try {
            // 1. 保存订单
            orderRepository.save(order);

            // 2. 创建本地消息
            LocalMessage message = new LocalMessage();
            message.setMessageId(UUID.randomUUID().toString());
            message.setMessage(JSON.toJSONString(order));
            message.setStatus("NEW");
            localMessageRepository.save(message);

            // 3. 提交事务
            transactionManager.commit(txStatus);

            // 4. 发送消息到Kafka
            kafkaTemplate.send("inventory-topic", message.getMessageId(), message.getMessage());
        } catch (Exception e) {
            // 回滚事务
            transactionManager.rollback(txStatus);
            throw new RuntimeException("Create order failed", e);
        }
    }
}

库存服务(Inventory Service)

@Service
public class InventoryService {

    @Autowired
    private InventoryRepository inventoryRepository;

    @KafkaListener(topics = "inventory-topic")
    public void handleOrderCreation(ConsumerRecord record) {
        String messageId = record.key();
        Order order = JSON.parseObject(record.value(), Order.class);

        try {
            // 扣减库存
            inventoryRepository.decreaseStock(order.getProductId(), order.getQuantity());

            // 确认消息处理成功
            kafkaTemplate.send("inventory-result-topic", messageId, "SUCCESS");
        } catch (Exception e) {
            // 消息处理失败，发送失败消息
            kafkaTemplate.send("inventory-result-topic", messageId, "FAILED");
        }
    }
}

本地消息表(Local Message Table)

@Entity
@Table(name = "local_message")
public class LocalMessage {
    @Id
    private String messageId;
    private String message;
    private String status; // NEW, SENT, CONFIRMED
    private Date createTime;
    private Date updateTime;

    // Getters and setters
}

消息恢复系统(Message Recovery System)

@Component
public class MessageRecoverySystem {

    @Autowired
    private LocalMessageRepository localMessageRepository;

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @Scheduled(fixedRate = 60000) // 每分钟执行一次
    public void recoverFailedMessages() {
        List failedMessages = localMessageRepository.findByStatusAndCreateTimeBefore("NEW", new Date(System.currentTimeMillis() - 300000)); // 5分钟前的消息

        for (LocalMessage message : failedMessages) {
            try {
                kafkaTemplate.send("inventory-topic", message.getMessageId(), message.getMessage());
                message.setStatus("SENT");
                localMessageRepository.save(message);
            } catch (Exception e) {
                // 记录日志，等待下次重试
                log.error("Failed to recover message: " + message.getMessageId(), e);
            }
        }
    }

    @KafkaListener(topics = "inventory-result-topic")
    public void handleInventoryResult(ConsumerRecord record) {
        String messageId = record.key();
        String result = record.value();

        LocalMessage message = localMessageRepository.findById(messageId).orElse(null);
        if (message != null) {
            if ("SUCCESS".equals(result)) {
                message.setStatus("CONFIRMED");
            } else {
                message.setStatus("FAILED");
            }
            localMessageRepository.save(message);
        }
    }
}