Java分布式事务案例

本文目录导读：

1. 分布式事务场景描述
2. 常见解决方案及实现
3. 配置文件示例
4. 消息表设计（可靠消息方案）
5. 最佳实践建议

我将为您提供一个完整的Java分布式事务案例分析,涵盖常见解决方案和实际代码示例。

分布式事务场景描述

假设我们有一个电商系统,包含以下服务：

• 订单服务：创建订单
• 库存服务：扣减库存
• 积分服务：增加用户积分

常见解决方案及实现

方案1：基于消息队列的最终一致性（推荐）

// 1. 订单服务 - 发送消息
@Service
public class OrderService {
    @Autowired
    private OrderMapper orderMapper;
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    @Transactional
    public void createOrder(OrderDO order) {
        // 本地事务：创建订单
        orderMapper.insert(order);
        // 发送消息到Kafka
        OrderMessage message = new OrderMessage();
        message.setOrderId(order.getId());
        message.setUserId(order.getUserId());
        message.setProductId(order.getProductId());
        message.setQuantity(order.getQuantity());
        kafkaTemplate.send("order-create-topic", 
            String.valueOf(order.getId()), message);
    }
}

// 2. 库存服务 - 消费消息
@Service
public class InventoryService {
    @Autowired
    private InventoryMapper inventoryMapper;
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    @KafkaListener(topics = "order-create-topic")
    @Transactional
    public void handleOrderCreate(OrderMessage message) {
        try {
            // 扣减库存
            int result = inventoryMapper.decreaseStock(
                message.getProductId(), 
                message.getQuantity());
            if (result > 0) {
                // 成功，发送成功消息
                InventoryResult resultMsg = new InventoryResult();
                resultMsg.setOrderId(message.getOrderId());
                resultMsg.setSuccess(true);
                kafkaTemplate.send("inventory-result-topic", 
                    String.valueOf(message.getOrderId()), resultMsg);
            } else {
                // 库存不足，发送失败消息
                sendFailure(message.getOrderId());
            }
        } catch (Exception e) {
            sendFailure(message.getOrderId());
            throw e; // 触发重试
        }
    }
    private void sendFailure(Long orderId) {
        InventoryResult resultMsg = new InventoryResult();
        resultMsg.setOrderId(orderId);
        resultMsg.setSuccess(false);
        kafkaTemplate.send("inventory-result-topic", 
            String.valueOf(orderId), resultMsg);
    }
}

// 3. 订单服务 - 处理结果
@Service
public class OrderCallbackService {
    @Autowired
    private OrderMapper orderMapper;
    @KafkaListener(topics = "inventory-result-topic")
    public void handleInventoryResult(InventoryResult result) {
        OrderDO order = orderMapper.selectById(result.getOrderId());
        if (result.isSuccess()) {
            // 更新订单状态为"已支付"
            orderMapper.updateStatus(result.getOrderId(), 
                OrderStatus.PAID.getCode());
        } else {
            // 更新订单状态为"已取消"
            orderMapper.updateStatus(result.getOrderId(), 
                OrderStatus.CANCELLED.getCode());
        }
    }
}

方案2：Seata AT模式（强一致性）

// 1. 引入Seata依赖
// pom.xml
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
</dependency>

// 2. 订单服务
@Service
public class OrderService {
    @Autowired
    private OrderMapper orderMapper;
    @Autowired
    private InventoryService inventoryService;
    @Autowired
    private PointService pointService;
    @GlobalTransactional(name = "create-order", rollbackFor = Exception.class)
    public void createOrder(OrderDO order) {
        // 创建订单
        orderMapper.insert(order);
        // 调用库存服务扣减库存
        inventoryService.decreaseStock(
            order.getProductId(), 
            order.getQuantity());
        // 调用积分服务增加积分
        pointService.addPoints(
            order.getUserId(), 
            order.getAmount().intValue() / 100); // 每消费100元得1积分
    }
}

// 3. 库存服务 - 使用@Transactional参与全局事务
@Service
public class InventoryService {
    @Autowired
    private InventoryMapper inventoryMapper;
    @Transactional
    public void decreaseStock(Long productId, Integer quantity) {
        // 扣减库存，如果失败会抛异常触发全局回滚
        int result = inventoryMapper.decreaseStock(productId, quantity);
        if (result == 0) {
            throw new RuntimeException("库存不足");
        }
    }
}

// 4. 积分服务
@Service
public class PointService {
    @Autowired
    private PointMapper pointMapper;
    @Transactional
    public void addPoints(Long userId, Integer points) {
        // 增加积分
        pointMapper.addPoints(userId, points);
    }
    }

方案3：TCC（Try-Confirm-Cancel）模式

// 1. 定义订单事务接口
public interface OrderTccService {
    @TwoPhaseBusinessAction(
        name = "createOrderTcc",
        commitMethod = "confirmOrder",
        rollbackMethod = "cancelOrder",
        source = "orderTccService"
    )
    void prepareCreateOrder(BusinessActionContext context, 
                           OrderDO order);
    boolean confirmOrder(BusinessActionContext context);
    boolean cancelOrder(BusinessActionContext context);
}

// 2. 实现订单事务
@Service
public class OrderTccServiceImpl implements OrderTccService {
    @Autowired
    private OrderMapper orderMapper;
    @Override
    public void prepareCreateOrder(BusinessActionContext context, 
                                  OrderDO order) {
        // Try阶段：创建订单（状态为"待确认"）
        order.setStatus(OrderStatus.PREPARE.getCode());
        orderMapper.insert(order);
        // 将订单ID存入上下文
        context.setActionContext("orderId", order.getId());
    }
    @Override
    public boolean confirmOrder(BusinessActionContext context) {
        // Confirm阶段：更新订单状态
        Long orderId = context.getActionContext("orderId");
        return orderMapper.updateStatus(orderId, 
            OrderStatus.CONFIRMED.getCode()) > 0;
    }
    @Override
    public boolean cancelOrder(BusinessActionContext context) {
        // Cancel阶段：更新订单状态
        Long orderId = context.getActionContext("orderId");
        return orderMapper.updateStatus(orderId, 
            OrderStatus.CANCELLED.getCode()) > 0;
    }
}

// 3. 库存服务 - TCC实现
@Service
public class InventoryTccService {
    @Autowired
    private InventoryMapper inventoryMapper;
    @TwoPhaseBusinessAction(stepName = "decreaseStock")
    public void tryDecreaseStock(BusinessActionContext context,
                                Long productId, 
                                Integer quantity) {
        // Try阶段：冻结库存
        inventoryMapper.freezeStock(productId, quantity);
    }
    public boolean confirmDecreaseStock(BusinessActionContext context) {
        // Confirm阶段：确认扣减
        Long productId = context.getActionContext("productId");
        Integer quantity = context.getActionContext("quantity");
        return inventoryMapper.confirmDecrease(productId, quantity);
    }
    public boolean cancelDecreaseStock(BusinessActionContext context) {
        // Cancel阶段：释放冻结库存
        Long productId = context.getActionContext("productId");
        Integer quantity = context.getActionContext("quantity");
        return inventoryMapper.unfreezeStock(productId, quantity);
    }
}

配置文件示例

# application.yml - Seata配置
seata:
  enabled: true
  application-id: order-service
  tx-service-group: my_transaction_group
  service:
    vgroup-mapping:
      my_transaction_group: default
    grouplist:
      default: 127.0.0.1:8091
  registry:
    type: nacos
    nacos:
      server-addr: 127.0.0.1:8848
      namespace: ""
      group: SEATA_GROUP

# application.yml - Kafka配置
spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: order-service-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

消息表设计（可靠消息方案）

-- 消息状态表
CREATE TABLE `transaction_message` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `message_id` varchar(64) NOT NULL COMMENT '消息ID',
  `biz_type` varchar(32) NOT NULL COMMENT '业务类型',
  `biz_id` varchar(64) NOT NULL COMMENT '业务ID',
  `message_body` text COMMENT '消息内容',
  `status` tinyint(4) NOT NULL DEFAULT '0' COMMENT '0-待发送 1-已发送 2-已消费 3-发送失败',
  `retry_times` int(11) DEFAULT '0' COMMENT '重试次数',
  `next_retry_time` datetime DEFAULT NULL COMMENT '下次重试时间',
  `create_time` datetime DEFAULT CURRENT_TIMESTAMP,
  `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  KEY `idx_status_next_retry` (`status`, `next_retry_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

最佳实践建议

方案选择指南

• 强一致性要求极高：使用Seata AT模式
• 性能和可用性优先：使用消息队列最终一致性
• 特殊业务场景：使用TCC模式

可靠性保障

• 消息表方式：将消息与本地事务一起提交
• 幂等处理：所有消费方必须实现幂等
• 死信队列：处理失败的消息
• 定时任务：定期扫描并重发未消费的消息

性能优化

• 批量处理消息
• 异步处理非关键环节
• 合理设置超时时间

监控报警

• 监控消息积压量
• 监控事务执行时间
• 设置合理的重试策略

这个案例涵盖了主流的分布式事务解决方案，您可以根据实际业务需求选择合适的方案，对于大多数电商场景，推荐使用消息队列实现的最终一致性方案,因为它平衡了性能和一致性。