开发者日记

外观

订单积分最终一致性：本地消息表实战与 Seata AT 实测对比 ​

问题背景：订单成功后，积分为什么会丢 ​

在订单系统中，用户支付成功后通常要执行两类操作：订单服务更新订单状态，积分服务给用户增加成长积分。它们位于不同数据库、不同服务实例，无法依赖一个本地事务同时提交。

常见失败场景很典型：订单状态已经提交，但调用积分服务超时；或者积分服务先加了积分，订单事务却回滚；更隐蔽的是调用实际成功，响应在网络中丢失，订单服务重试后又加了一次积分。此时强一致性、高性能、可用性三者无法同时占满，业务通常需要根据场景选择一致性方案。

本文以“订单支付成功后发放积分”为背景，使用 Java 17、Spring Boot 3、Spring JDBC、MySQL 8 实现一套可运行的本地消息表方案，并与 Seata AT 模式进行架构和工程成本对比。

方案对比：本地消息表、事务消息与 Seata AT ​

本地消息表的核心思想是：业务数据与待发送消息放在同一个本地事务中提交，然后由异步任务扫描消息并投递到 MQ 或直接调用下游。下游通过业务唯一键保证幂等，失败消息不断重试，从而达到最终一致性。

它和 RocketMQ 事务消息、Seata AT 的差异如下：

方案一致性机制基础设施要求适用场景主要代价
本地消息表本地事务 + 异步重试 + 消费幂等普通数据库、定时任务、MQ 可选订单、积分、通知等可容忍短暂延迟的业务需要维护消息状态、补偿和清理
RocketMQ 事务消息Half Message + 事务回查依赖 RocketMQ消息投递能力成熟的异步场景中间件绑定较强，仍需消费幂等
Seata AT两阶段提交 + undo_log 自动补偿TC、RM、代理数据源、全局锁短事务、强一致诉求高、模型简单的跨库操作性能损耗、全局锁、脏写与兼容性问题

对于订单积分，业务一般允许积分在几百毫秒或几秒内到账，而且“加积分”天然适合幂等处理，因此本地消息表是成本较低、稳定性较高的方案。

支付成功事件订单服务接收订单库本地事务更新订单 + 插入消息消息投递器扫描与重试积分服务幂等入账失败进入退避重试次数、错误、下次执行成功标记 DONE记录完成时间

数据库设计：业务单据与本地消息同库同事务 ​

首先建立订单表、本地消息表，以及积分服务的幂等流水表。重点不是表有多少字段，而是消息状态、唯一键、重试字段和状态跳转条件必须完整。

sql
-- 订单库：订单表
CREATE TABLE t_order (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  order_no VARCHAR(64) NOT NULL,
  user_id BIGINT NOT NULL,
  status VARCHAR(32) NOT NULL COMMENT 'INIT/PAID/CANCELLED',
  pay_amount BIGINT NOT NULL COMMENT '支付金额，单位分',
  updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  UNIQUE KEY uk_order_no (order_no)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 订单库：本地消息表，必须与业务表在同一个数据库实例
CREATE TABLE t_local_message (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  message_id VARCHAR(64) NOT NULL COMMENT '消息唯一 ID，建议 UUID',
  biz_key VARCHAR(128) NOT NULL COMMENT '业务幂等键，如 ORDER_POINT:订单号',
  topic VARCHAR(64) NOT NULL,
  payload JSON NOT NULL,
  status VARCHAR(16) NOT NULL COMMENT 'INIT/SENDING/DONE/DEAD',
  retry_count INT NOT NULL DEFAULT 0,
  max_retry INT NOT NULL DEFAULT 20,
  next_retry_at DATETIME NOT NULL,
  last_error VARCHAR(512) NULL,
  created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  UNIQUE KEY uk_message_id (message_id),
  KEY idx_status_next_retry (status, next_retry_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 积分库：幂等入账流水
CREATE TABLE t_point_ledger (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  idempotent_key VARCHAR(128) NOT NULL,
  user_id BIGINT NOT NULL,
  points INT NOT NULL,
  order_no VARCHAR(64) NOT NULL,
  created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  UNIQUE KEY uk_idempotent_key (idempotent_key)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

t_local_message 与订单表在同一个库内，才能被同一个数据库事务覆盖。不要把消息先写入 Redis、文件或另一个服务，否则“业务成功、消息丢失”的问题仍然存在。

Java 实现：支付成功时原子写入订单和消息 ​

下面代码省略 Controller，只保留核心服务。订单状态更新和消息插入必须在同一个 @Transactional 方法中完成。

java
package com.demo.order;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.Map;
import java.util.UUID;

@Service
public class OrderPayService {

    private final JdbcTemplate jdbcTemplate;
    private final ObjectMapper objectMapper;

    public OrderPayService(JdbcTemplate jdbcTemplate, ObjectMapper objectMapper) {
        this.jdbcTemplate = jdbcTemplate;
        this.objectMapper = objectMapper;
    }

    @Transactional
    public void markOrderPaid(String orderNo, Long userId, Long payAmount) {
        // 1. 使用条件更新避免重复支付通知把订单状态改乱
        int rows = jdbcTemplate.update("""
            UPDATE t_order
               SET status = 'PAID'
             WHERE order_no = ? AND status = 'INIT'
            """, orderNo);

        if (rows == 0) {
            // 生产环境可按业务返回幂等成功，这里直接抛出以便观察重复请求
            throw new IllegalStateException(
                "订单不存在或状态不是 INIT，orderNo=" + orderNo);
        }

        // 2. 构造积分消息，1 元 = 1 积分，按分换算并四舍五入由业务规则决定
        int points = (int) Math.max(1L, payAmount / 100);
        Map<String, Object> payload = Map.of(
            "orderNo", orderNo,
            "userId", userId,
            "points", points,
            "event", "ORDER_PAID_GRANT_POINT"
        );

        try {
            String messageId = UUID.randomUUID().toString();
            String bizKey = "ORDER_POINT:" + orderNo;

            // 3. 消息与订单状态在同一个本地事务中提交
            jdbcTemplate.update("""
                INSERT INTO t_local_message(
                    message_id, biz_key, topic, payload, status,
                    retry_count, max_retry, next_retry_at
                ) VALUES (?, ?, ?, ?, 'INIT', 0, 20, NOW())
                """,
                messageId,
                bizKey,
                "POINT_GRANT",
                objectMapper.writeValueAsString(payload)
            );
        } catch (JsonProcessingException e) {
            throw new IllegalArgumentException("积分消息序列化失败", e);
        }
    }
}

这里有一个关键细节：消息 ID 可以使用 UUID，但消费幂等不能只依赖消息 ID。因为运营补单、人工重放、消息重建时可能生成新的消息 ID，积分服务更应认业务键 ORDER_POINT:订单号。

消息投递器：加锁认领、状态推进与退避重试 ​

单节点可以直接定时扫描；多节点部署时必须先认领任务，避免多个实例重复投递。MySQL 可以使用条件更新实现轻量任务锁。

java
package com.demo.order;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.client.RestClient;

import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;

@Component
public class LocalMessageDispatcher {

    private static final int BATCH_SIZE = 100;
    private final JdbcTemplate jdbcTemplate;
    private final ObjectMapper objectMapper;
    private final RestClient restClient;

    public LocalMessageDispatcher(JdbcTemplate jdbcTemplate,
                                  ObjectMapper objectMapper,
                                  RestClient.Builder builder) {
        this.jdbcTemplate = jdbcTemplate;
        this.objectMapper = objectMapper;
        this.restClient = builder.baseUrl("http://point-service").build();
    }

    @Scheduled(fixedDelay = 1000)
    public void dispatch() {
        List<Long> ids = jdbcTemplate.queryForList("""
            SELECT id FROM t_local_message
             WHERE status IN ('INIT', 'SENDING')
               AND next_retry_at <= NOW()
             ORDER BY id ASC
             LIMIT ?
            """, Long.class, BATCH_SIZE);

        for (Long id) {
            dispatchOne(id);
        }
    }

    @Transactional
    public void dispatchOne(Long id) {
        // 条件更新充当数据库行级认领，防止多实例并发发送同一条消息
        int locked = jdbcTemplate.update("""
            UPDATE t_local_message
               SET status = 'SENDING', updated_at = NOW()
             WHERE id = ? AND status IN ('INIT', 'SENDING')
            """, id);
        if (locked == 0) {
            return;
        }

        Map<String, Object> row = jdbcTemplate.queryForMap(
            "SELECT message_id, biz_key, payload, retry_count, max_retry " +
            "FROM t_local_message WHERE id = ?", id);

        try {
            JsonNode payload = objectMapper.readTree(row.get("payload").toString());

            // 实际项目建议投递 MQ；这里直接 HTTP 调用，便于示例完整运行
            String response = restClient.post()
                .uri("/points/grant")
                .header("X-Message-Id", row.get("message_id").toString())
                .body(Map.of(
                    "idempotentKey", row.get("biz_key"),
                    "userId", payload.get("userId").asLong(),
                    "points", payload.get("points").asInt(),
                    "orderNo", payload.get("orderNo").asText()
                ))
                .retrieve()
                .body(String.class);

            if (!"OK".equals(response)) {
                throw new IllegalStateException("积分服务返回非 OK: " + response);
            }

            jdbcTemplate.update("""
                UPDATE t_local_message
                   SET status = 'DONE', last_error = NULL, updated_at = NOW()
                 WHERE id = ?
                """, id);

        } catch (Exception ex) {
            markRetryOrDead(id, (Integer) row.get("retry_count"),
                            (Integer) row.get("max_retry"), ex);
        }
    }

    private void markRetryOrDead(Long id, int retryCount,
                                 int maxRetry, Exception ex) {
        int nextRetry = retryCount + 1;
        String error = ex.getMessage() == null ? ex.getClass().getSimpleName() : ex.getMessage();
        error = error.length() > 500 ? error.substring(0, 500) : error;

        if (nextRetry >= maxRetry) {
            jdbcTemplate.update("""
                UPDATE t_local_message
                   SET status = 'DEAD', retry_count = ?,
                       last_error = ?, updated_at = NOW()
                 WHERE id = ?
                """, nextRetry, error, id);
            // 生产环境应发送告警并接入人工补偿台，而不是只写死信状态
            return;
        }

        // 指数退避：10s、30s、70s...，避免下游故障时被重试流量压垮
        LocalDateTime nextTime = LocalDateTime.now()
            .plusSeconds(10L * (1L << Math.min(nextRetry, 5)));

        jdbcTemplate.update("""
            UPDATE t_local_message
               SET status = 'INIT', retry_count = ?,
                   last_error = ?, next_retry_at = ?, updated_at = NOW()
             WHERE id = ?
            """, nextRetry, error, nextTime, id);
    }
}

启动类还需要增加 @EnableScheduling。如果投递目标是 MQ，发送成功确认后再把状态改为 DONE；消费者处理失败时由 MQ 重试，本地消息表只负责把消息可靠送达到 MQ。

积分服务幂等入账：唯一索引是最后一道防线 ​

即使投递端发生超时重试，积分服务也必须保证同一业务键只入账一次。

java
package com.demo.point;

import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Map;

@Service
public class PointGrantService {

    private final JdbcTemplate jdbcTemplate;

    public PointGrantService(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }

    @Transactional
    public String grant(Map<String, Object> request) {
        String key = request.get("idempotentKey").toString();
        Long userId = Long.valueOf(request.get("userId").toString());
        Integer points = Integer.valueOf(request.get("points").toString());
        String orderNo = request.get("orderNo").toString();

        try {
            // 先写流水，唯一索引保证同一订单不会重复发积分
            jdbcTemplate.update("""
                INSERT INTO t_point_ledger(idempotent_key, user_id, points, order_no)
                VALUES (?, ?, ?, ?)
                """, key, userId, points, orderNo);
        } catch (DuplicateKeyException e) {
            // 重复请求视为成功，让投递端可以结束补偿
            return "OK";
        }

        jdbcTemplate.update("""
            UPDATE t_user_point
               SET total_points = total_points + ?
             WHERE user_id = ?
            """, points, userId);

        return "OK";
    }
}

如果担心插入流水成功但更新总积分失败，可以把流水和总积分放在积分库同一个本地事务中。若总积分表数据量很大，也可以只保留流水，通过异步聚合生成分数，但查询链路要相应改造。

与 Seata AT 的差异：自动化背后的成本 ​

Seata AT 通过拦截数据库操作，在一阶段提交业务数据前后生成 undo_log，并由 TC 协调二阶段提交或回滚。它对业务代码侵入较低，一个 @GlobalTransactional 即可覆盖订单和积分服务。

java
@GlobalTransactional
public void paySuccess(String orderNo) {
    orderMapper.markPaid(orderNo);
    pointClient.grant(orderNo); // 异常时 Seata 根据 undo_log 反向补偿
}

但 AT 模式并不是“无成本分布式事务”。第一，它要求每个参与库都有 undo_log 表，并使用受支持的 JDBC 代理方式；第二，一阶段会持有全局锁，热点订单并发更新时可能明显降低吞吐；第三，AT 默认依赖本地事务隔离和全局锁读一致性，跨服务中若混入不支持自动补偿的外部接口，仍需要人工设计幂等和兜底；第四，TC 成为关键组件，需要考虑高可用、容量、超时和故障恢复。

在订单积分这种天然异步、可重试、只需要最终一致的场景中，本地消息表通常更合适。Seata AT 更适合跨库写操作必须同成同败、事务持续时间短、并发冲突可控，并且团队能接受中间件和全局锁成本的内部系统。

避坑总结：生产环境最容易踩的 5 个问题 ​

  1. 业务和消息不在同一个本地事务中提交。 常见错误是先提交订单，再调用 messageService.save()，一旦应用在两步之间重启，积分永久丢失。修正方法是保证订单更新与消息插入使用同一个数据源、同一个事务，不要经过异步线程或远程服务写消息。

  2. 消费端只判断消息 ID，不判断业务唯一键。 现象是 MQ 重试没有重复，但人工补单或消息重建后用户收到两次积分。修正方法是积分流水使用 ORDER_POINT:订单号 建唯一索引，消息 ID 只用于链路追踪。

  3. 失败后无限立即重试。 下游数据库抖动时，大量消息每秒重试，形成雪崩。修正方法是记录 retry_count 和 next_retry_at，使用指数退避并设置最大重试次数，超过阈值进入 DEAD 并告警。

  4. 多实例调度没有认领机制。 现象是日志中同一条消息被两个节点同时发送。虽然幂等键能挡住重复入账，但会放大下游压力。修正方法是通过条件更新、SELECT ... FOR UPDATE SKIP LOCKED 或调度分片保证同一消息只被一个执行器处理。

  5. 把 Seata AT 当成所有外部调用的万能回滚。 调用短信、支付、第三方券码等接口后，即使本地事务回滚，外部动作也未必能撤销。修正方法是将这类动作放到事务提交后的可靠事件中，并为外部系统设计取消接口或状态查询接口。

本地消息表不是强一致性魔法，它把复杂的跨库事务拆成“本地原子写入、可靠投递、幂等消费、失败补偿”四个可控环节。订单积分场景下，只要消息可追踪、消费可重入、死信可告警，系统就能在不引入重型全局事务协调器的情况下，稳定达到最终一致性。

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。

本文来自开发者日记（devdiary.cn）· 记录开发者实战笔记与免费在线工具箱

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。