0
0
0

分布式事务解决方案：最佳实践

孙涛○
2025-06-17 12:16
阅读 494

开篇：什么是分布式事务，为什么我们需要它？

开篇：什么是分布式事务，为什么我们需要它？

想象一下你在网上买一件衣服。当你下单后，系统需要做两件事：

  1. 从你的账户扣钱
  2. 把这件衣服的状态标记为“已售出”

如果这两步操作不是同步完成的会怎样？比如，钱被扣了，但商品没被标记成已卖出，或者反过来？

这就会带来数据不一致的问题。

在单体应用中，我们使用数据库的**本地事务（Local Transaction）**来保证这两个操作要么都成功、要么都失败。然而，当我们的系统变得越来越复杂，拆分成了多个独立的服务（比如订单服务、库存服务、支付服务），这些服务通常各自连接不同的数据库。这就引出了一个新问题——如何在多个服务之间保持事务的一致性？

这个问题就是**分布式事务（Distributed Transaction）**要解决的核心。

常见场景举例

  • 银行转账：从A账户转钱到B账户，涉及两个不同银行的数据更新
  • 网购下单：扣除库存并支付，可能跨库存服务和支付服务
  • 在线订票：锁定座位+支付，可能调用两个不同的业务系统

我们的目标

在这篇文章中，我们将一步步学习：

  • 如何理解分布式事务的基本概念
  • 几种主流的分布式事务解决方案
  • 每种方案的工作原理及适用场景
  • 使用代码示例演示具体实现
  • 新手常见的疑问与解答
  • 你可以继续深入学习的方向

这篇文章专为完全零基础的初学者设计，没有晦涩的专业术语堆砌，而是通过最通俗易懂的语言 + 具体代码案例的方式来讲解这个听起来很高大上的主题。

环境准备：搭建开发环境

环境准备：搭建开发环境

为了更好地理解分布式事务的实现方式，我们要先准备好本地的开发环境。我们将使用 Java + Spring Boot + MySQL + RocketMQ 来构建一个简单的项目。

所需工具

工具名称 版本推荐 安装说明
JDK 17或以上 推荐使用 OpenJDK
IntelliJ IDEA 社区版即可 官网下载安装
Maven 最新版本 自动集成
MySQL 8.0+ 安装时记得设置 root 密码
RocketMQ 4.9+ 本地可以使用简单模式启动
Postman 用于接口测试 可选

步骤一：安装 JDK 和配置环境变量

略（网上教程很多）

步骤二：安装 IntelliJ IDEA

访问 https://www.jetbrains.com/idea/download/ 下载社区版并安装。

步骤三：安装 MySQL

可参考菜鸟教程或掘金文章《MySQL Windows/Mac 安装指南》。

创建两个库：

CREATE DATABASE order_service;
CREATE DATABASE inventory_service;

步骤四：安装 RocketMQ

RocketMQ 是我们在实战中用于消息驱动实现最终一致性的核心组件。

  1. 访问官网下载地址：https://rocketmq.apache.org/releases/#source-releases
  2. 解压后进入目录
  3. 启动 namesrv：
bin/mqnamesrv
  1. 再启动 broker：
bin/mqbroker -n localhost:9876 autoCreateTopicEnable=true

如果你遇到内存不足的问题，在 runserver.sh 或 runbroker.sh 中调整 JVM 参数即可。

步骤五：创建项目结构

打开 IntelliJ IDEA，新建一个 Spring Boot 项目：

  • 名称：distributed-transaction-demo
  • 包名：com.example.transaction.demo

添加以下依赖（pom.xml）：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.2.3</version>
    </dependency>
</dependencies>

别忘了在 application.yml 中配置两个数据库：

spring:
  datasource:
    url: jdbc:mysql://localhost:3306/order_service?useSSL=false&serverTimezone=UTC
    username: root
    password: your_password
    driver-class-name: com.mysql.cj.jdbc.Driver

这样我们就可以开始编码啦！

核心概念：搞懂关键术语

核心概念：搞懂关键术语

在正式动手前，让我们先了解几个最重要的概念。

1. 本地事务（Local Transaction）

指的是在单一数据库内的操作，可以通过数据库的事务机制（BEGIN / COMMIT / ROLLBACK）来控制。

例如：

START TRANSACTION;
UPDATE account SET balance = balance - 100 WHERE id = 1;
UPDATE goods SET stock = stock - 1 WHERE id = 100;
COMMIT;

如果其中任何一句失败，整个事务回滚，不会留下错误状态。

2. 分布式事务（Distributed Transaction）

发生在多个数据库、多个服务之间的事务。这时，无法直接用本地事务机制来保障一致性，需要引入外部的协调机制。

3. CAP 定理

CAP 是关于分布式系统的经典理论：

  • Consistency（一致性）
  • Availability（可用性）
  • Partition Tolerance（分区容忍）

它们三者只能同时满足两点，不能全部兼顾。这是我们在选择分布式事务方案时的重要理论依据。

4. BASE 理论

BASE 是对 ACID 的补充，适用于分布式系统：

  • Basically Available（基本可用）
  • Soft state（柔性状态）
  • Eventually consistent（最终一致）

换句话说：我们可以接受短暂的数据不一致，只要最终能自动修复。

这也是大多数高并发互联网系统的处理方式。

实战项目：跟着做一次完整的练习

系统架构设计图-1

场景描述

我们来模拟一个电商下单场景：

  • 用户下订单时，需要：
    1. 创建订单记录（order_service）
    2. 扣除库存（inventory_service）

这两个操作要确保要么都成功，要么都不执行。

如果不一致，可能出现：

  • 钱扣了但商品未出售
  • 商品卖出去了但没收到钱

因此我们需要一个合适的分布式事务方案。

方案选择：基于 RocketMQ 的最终一致性

我们选择的是异步消息驱动 + 补偿机制，这是一种典型的最终一致方案，适合不要求强实时一致性的场景。

总体流程图如下：

用户下单
│
├─ 写入订单数据库 (order_service)
├─ 发送扣除库存消息 (RocketMQ)
├─ 库存服务监听 MQ 消息并扣除库存
└─ 如果失败，触发补偿逻辑（如重试、人工介入等）

现在我们一步一步实现它。

第一步：创建订单服务（order_service）

创建一个表：

USE order_service;

CREATE TABLE orders (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    product_id BIGINT NOT NULL,
    amount INT NOT NULL,
    user_id BIGINT NOT NULL
);

Spring Boot 对应的实体类：

@Entity
public class Order {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    private Long productId;
    private Integer amount;
    private Long userId;

    // getter/setter
}

Repository 接口：

public interface OrderRepository extends JpaRepository<Order, Long> {}

OrderController：

@RestController
@RequestMapping("/orders")
public class OrderController {

    @Autowired
    private OrderRepository orderRepository;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @PostMapping
    public String createOrder(@RequestBody Map<String, Object> payload) {
        Long productId = Long.valueOf((Integer) payload.get("productId"));
        Long userId = Long.valueOf((Integer) payload.get("userId"));

        Order order = new Order();
        order.setProductId(productId);
        order.setUserId(userId);
        order.setAmount(1);

        orderRepository.save(order);

        // 发送消息给库存服务
        Message<String> message = MessageBuilder.withPayload("deduct:" + productId).build();
        rocketMQTemplate.send("INVENTORY_TOPIC", message);

        return "订单创建成功，请等待库存扣除";
    }
}

第二步：库存服务（inventory_service）

创建另一个数据库，并创建产品库存表：

USE inventory_service;

CREATE TABLE products (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    name VARCHAR(255),
    stock INT DEFAULT 0
);

插入一条数据作为测试：

INSERT INTO products (name, stock) VALUES ('测试商品', 10);

Spring Boot 对应实体类：

@Entity
public class Product {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private String name;
    private Integer stock;
    
    // getter/setter
}

ProductRepository：

public interface ProductRepository extends JpaRepository<Product, Long> {}

库存消费者（监听 RocketMQ）：

@Component
@RocketMQMessageListener(topic = "INVENTORY_TOPIC", consumerGroup = "inventory-consumer-group")
public class InventoryConsumer implements RocketMQListener<MessageExt> {

    @Autowired
    private ProductRepository productRepository;

    @Override
    public void onMessage(MessageExt messageExt) {
        String body = new String(messageExt.getBody());
        if (body.startsWith("deduct:")) {
            Long productId = Long.valueOf(body.split(":")[1]);
            Product product = productRepository.findById(productId).orElse(null);
            if (product != null && product.getStock() > 0) {
                product.setStock(product.getStock() - 1);
                productRepository.save(product);
                System.out.println("库存扣除成功");
            } else {
                System.out.println("库存扣除失败，商品不存在或库存不足");
                // TODO: 这里可以加入失败重试机制或通知人工处理
            }
        }
    }
}

第三步：测试流程

  1. 启动 RocketMQ（先开 namesrv，再开 broker）
  2. 启动 Spring Boot 应用（订单服务 + 库存服务）
  3. 使用 Postman 发送 POST 请求到 http://localhost:8080/orders

Body 内容如下：

{
  "productId": 1,
  "userId": 123
}
  1. 观察控制台输出和数据库变化，确认是否发生异常

小结

我们实现了这样一个流程：

  • 用户下单 → 订单落库 → 异步发消息
  • 消息队列由库存服务消费 → 扣除库存
  • 若过程中任一环节失败 → 日志记录，后续可补发或人工处理

虽然不能做到严格一致，但通过异步方式实现了最终一致性，非常适合高并发的互联网场景。

常见问题解答

Q：为什么不用分布式事务中间件（如 Seata、XA）？

A：本文旨在以最轻量的方式入门分布式事务思想。Seata 等方案确实更强大，但部署复杂、学习曲线陡峭。建议你在掌握了基本思路后再尝试。

Q：如果 MQ 发送失败怎么办？

A：可以在订单服务发送消息前先写一个本地消息表，标记该消息是否已发送成功。定时任务检测未发送的消息进行重发。

Q：如果库存服务消费失败，会不会导致永久不一致？

A：是的，这就是最终一致性的代价。实际生产中我们会配合重试机制、死信队列、人工审核等方式来兜底。

Q：本地事务还能用吗？

A：当然可以！每个服务内部依然可以用本地事务来保障单库一致性。分布式事务是在“服务间”层面解决问题。

Q：能不能在订单服务直接调用库存服务的 API？

A：理论上是可以的（称为两阶段提交模型）。但在高并发下性能差、耦合度高。我们上面使用的 MQ 解耦是一个更现代的做法。

学习建议：下一步怎么学？

恭喜你完成了第一个完整的分布式事务实践项目！

下面是一些进阶方向供你选择：

1. 学习 Seata 实现 AT 模式

Seata 是阿里巴巴开源的分布式事务框架，AT 模式支持自动生成 undo_log 回滚日志，适合想快速实现分布式事务的项目。

2. 学习 Saga 模式与 TCC 模式

  • Saga 模式：将长事务拆分为多个子事务，失败后逐个回滚。
  • TCC 模式：Try-Confirm-Cancel 模式，适合对一致性要求较高的金融级交易系统。

3. 了解分布式锁的使用场景

有时候我们需要加锁来防止并发操作冲突，比如库存扣减。

  • Redis 分布式锁
  • Zookeeper 分布式锁
  • Etcd 锁服务

4. 学习事件溯源（Event Sourcing）与 CQRS 架构

这些属于更高级别的分布式架构设计，适合未来向架构师发展的同学。

结语

分布式事务听上去很复杂，其实它的本质非常简单：

“我要在多个地方操作数据，希望它们保持一致。”

我们选择了以 RocketMQ + 异步消息驱动的方式实现了这个目标，这正是互联网中最常用的思路之一。

记住一句话：

一致性永远是个权衡的过程，没有银弹，只有最合适的选择。

接下来你可以继续拓展 Seata、TCC、Saga 模式等更多内容，相信你会越来越熟练掌握这个强大的技能！

全文约 3871 字，完整阅读预计需 12~15 分钟

祝你学习愉快，有任何问题欢迎留言讨论！

标签:工具Go算法

评论 0

最热最新
暂无评论

为你推荐

暂无相关推荐
匿名用户Lv.1
0
影响力
0
文章
0
粉丝

相关推荐

暂无相关推荐

精选内容

暂无精选内容