最近在技术社区看到一个很有意思的现象一个看似“无厘头”的标题背后却可能隐藏着开发者对数据一致性、状态管理和异常处理的深刻困惑。标题里提到的“未退”、“收益折半”、“突然躺中”像极了我们在处理异步任务、金融计算或分布式事务时因为一个状态没更新、一个锁没释放或者一次幂等性没做好导致的数据“灵异事件”。这绝不是小孩的恶作剧而是系统在“恶作剧”开发者。今天我们就来彻底拆解这类问题。我将从一个高并发场景下的“收益计算异常”案例入手带你理解事务、幂等性和最终一致性的核心概念并用一套可落地的Spring Boot Redis MySQL方案演示如何从代码层面杜绝“收益折半”、“状态未退”的坑。如果你正在开发涉及支付、积分、任务奖励等有状态变化的系统这篇文章值得你仔细读完。1. 从“收益折半”案例看状态同步的核心痛点我们先还原一下标题中可能描述的场景一个用户“宝宝”完成了一个动作比如观看直播、完成任务预期获得0.1的收益“软米”。但系统实际只发放了0.05收益“折半”。同时用户可能发起了“退款”或“撤销”操作但系统状态显示“未退”仿佛被“涮”了。在技术层面这通常指向几个经典问题非原子性操作 “收益发放”不是一个原子操作。它可能被拆分为“计算收益”、“更新账户余额”、“记录流水”等多个步骤。如果系统在中间步骤崩溃或并发冲突就会导致状态不一致如流水记了余额没加。并发更新丢失 高并发下两个线程同时读取用户余额假设为0都加上0.1然后先后写回结果余额变成了0.1而不是0.2。这就是“更新丢失”感觉收益被“吞”了一半。缺乏幂等性 用户请求可能因为网络超时被客户端重试如果服务端没有识别出这是同一个请求就会重复处理导致收益被发放两次或者“退款”操作执行两次。但有时因为并发和状态判断错误可能只成功了一次看起来像“折半”或“未退”。状态机设计混乱 “已发放”、“发放中”、“已退款”、“退款中”这些状态如果没有清晰的定义和严格的转换规则就容易出现“未退”但钱已扣或者“已退”但状态没更新的情况。本文的核心判断是大多数这类“灵异”问题根源不在于复杂的算法而在于对“事务边界”、“并发控制”和“状态管理”这些基础机制的理解不到位或实现有瑕疵。接下来我们将通过一个模拟的“任务奖励发放系统”从原理到实战构建一个健壮的解决方案。2. 核心概念事务、幂等性与最终一致性在深入代码之前必须厘清三个基石概念。2.1 数据库事务 (ACID)事务是保证单个数据库内数据操作“要么全做要么全不做”的机制。其ACID特性原子性 (Atomicity) 事务内的操作是一个不可分割的整体。这正是解决“收益折半”的第一步。如果发放收益涉及更新用户表和插入流水表它们必须在同一个事务中。一致性 (Consistency) 事务执行前后数据库必须保持一致性状态如余额不能为负。隔离性 (Isolation) 并发事务之间相互隔离防止脏读、不可重复读、幻读。隔离级别如读已提交、可重复读的选择直接影响并发问题。持久性 (Durability) 事务提交后对数据的修改是永久性的。在分布式环境下单纯依赖数据库事务往往不够因为我们的服务、缓存、消息队列可能不在同一个数据库实例上。2.2 幂等性 (Idempotence)这是一个数学概念在计算机科学中指无论操作执行一次还是多次其产生的结果都是相同的。对于HTTP接口意味着客户端用同样的参数重复调用服务端的状态变化只发生一次。为什么需要幂等网络是不稳定的。用户点击“领取奖励”按钮可能因为响应慢而多次点击客户端超时后自动重试消息队列消费者失败后重新投递。如果没有幂等控制就会导致重复发放。实现幂等性的常见手段 唯一业务流水号、数据库唯一索引、Token机制、状态机判断。2.3 最终一致性 (Eventual Consistency)这是分布式系统领域CAP理论中的一种一致性模型。它不要求数据时刻保持强一致但保证在没有新的更新操作后经过一段时间所有副本的数据最终会达到一致状态。与我们场景的关联 当“发放收益”需要调用外部积分系统、会计系统时我们可能采用“先更新本地数据库状态为‘发放中’然后异步调用外部系统成功后更新为‘已发放’”的模式。这期间用户查询到的状态可能是“发放中”但外部系统可能已处理完成这就是短暂的“不一致”。我们需要通过补偿机制如对账、重试来达到最终一致。3. 环境准备与项目初始化我们将创建一个Spring Boot项目来模拟奖励发放系统。技术栈JDK: 17 或 21Spring Boot: 3.1.x 或 3.2.x持久层: Spring Data JPA MySQL 8.0缓存与分布式锁: Spring Data Redis Redisson构建工具: Maven数据库表设计user_account用户账户表存储余额。reward_task奖励任务表定义任务和奖励金额。reward_record奖励发放记录表核心表用于实现幂等和状态追踪。account_flow账户流水表记录每一笔变动。使用schema.sql初始化数据库-- schema.sql CREATE DATABASE IF NOT EXISTS reward_system DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; USE reward_system; -- 用户账户表 CREATE TABLE user_account ( id BIGINT PRIMARY KEY AUTO_INCREMENT COMMENT 主键, user_id VARCHAR(64) NOT NULL UNIQUE COMMENT 用户唯一标识, balance DECIMAL(15, 2) NOT NULL DEFAULT 0.00 COMMENT 账户余额, version INT NOT NULL DEFAULT 0 COMMENT 乐观锁版本号, created_time DATETIME DEFAULT CURRENT_TIMESTAMP, updated_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_user_id (user_id) ) COMMENT 用户账户; -- 奖励任务表 CREATE TABLE reward_task ( id BIGINT PRIMARY KEY AUTO_INCREMENT, task_code VARCHAR(64) NOT NULL UNIQUE COMMENT 任务编码, task_name VARCHAR(255) NOT NULL COMMENT 任务名称, reward_amount DECIMAL(10, 2) NOT NULL COMMENT 奖励金额, is_active BOOLEAN DEFAULT TRUE COMMENT 是否生效, created_time DATETIME DEFAULT CURRENT_TIMESTAMP ) COMMENT 奖励任务定义; -- 奖励发放记录表 (幂等性核心) CREATE TABLE reward_record ( id BIGINT PRIMARY KEY AUTO_INCREMENT, biz_no VARCHAR(128) NOT NULL UNIQUE COMMENT 业务唯一流水号(用于幂等), user_id VARCHAR(64) NOT NULL COMMENT 用户ID, task_id BIGINT NOT NULL COMMENT 任务ID, reward_amount DECIMAL(10, 2) NOT NULL COMMENT 奖励金额, status TINYINT NOT NULL COMMENT 状态: 0-处理中, 1-发放成功, 2-发放失败, 3-已撤销, retry_count INT DEFAULT 0 COMMENT 重试次数, error_msg TEXT COMMENT 错误信息, created_time DATETIME DEFAULT CURRENT_TIMESTAMP, updated_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_biz_no (biz_no), INDEX idx_user_task (user_id, task_id), INDEX idx_status_created (status, created_time) ) COMMENT 奖励发放记录; -- 账户流水表 CREATE TABLE account_flow ( id BIGINT PRIMARY KEY AUTO_INCREMENT, flow_no VARCHAR(128) NOT NULL UNIQUE COMMENT 流水号, user_id VARCHAR(64) NOT NULL COMMENT 用户ID, amount DECIMAL(10, 2) NOT NULL COMMENT 变动金额正为入账负为出账, balance_before DECIMAL(15, 2) NOT NULL COMMENT 变动前余额, balance_after DECIMAL(15, 2) NOT NULL COMMENT 变动后余额, biz_type VARCHAR(32) NOT NULL COMMENT 业务类型: REWARD-任务奖励, REFUND-退款, biz_no VARCHAR(128) NOT NULL COMMENT 关联业务号(如reward_record.biz_no), remark VARCHAR(512) COMMENT 备注, created_time DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_user_id (user_id), INDEX idx_biz_no (biz_no), INDEX idx_created_time (created_time) ) COMMENT 账户流水;Maven依赖 (pom.xml):?xml version1.0 encodingUTF-8? project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instance xsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd modelVersion4.0.0/modelVersion parent groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-parent/artifactId version3.1.6/version !-- 请根据实际情况调整 -- relativePath/ /parent groupIdcom.example/groupId artifactIdreward-system/artifactId version0.0.1-SNAPSHOT/version namereward-system/name descriptionDemo project for reward distribution/description properties java.version17/java.version /properties dependencies !-- Web -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency !-- JPA -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-data-jpa/artifactId /dependency !-- MySQL Driver -- dependency groupIdcom.mysql/groupId artifactIdmysql-connector-j/artifactId scoperuntime/scope /dependency !-- Redis -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-data-redis/artifactId /dependency !-- Redisson for distributed lock -- dependency groupIdorg.redisson/groupId artifactIdredisson-spring-boot-starter/artifactId version3.23.5/version !-- 请检查最新版本 -- /dependency !-- Lombok -- dependency groupIdorg.projectlombok/groupId artifactIdlombok/artifactId optionaltrue/optional /dependency !-- Test -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-test/artifactId scopetest/scope /dependency /dependencies build plugins plugin groupIdorg.springframework.boot/groupId artifactIdspring-boot-maven-plugin/artifactId configuration excludes exclude groupIdorg.projectlombok/groupId artifactIdlombok/artifactId /exclude /excludes /configuration /plugin /plugins /build /project应用配置文件 (application.yml):# application.yml spring: datasource: url: jdbc:mysql://localhost:3306/reward_system?useUnicodetruecharacterEncodingutf8useSSLfalseserverTimezoneAsia/Shanghai username: root password: your_password driver-class-name: com.mysql.cj.jdbc.Driver jpa: hibernate: ddl-auto: update # 首次启动可设为update创建表生产环境建议使用none通过sql脚本管理 show-sql: true properties: hibernate: format_sql: true dialect: org.hibernate.dialect.MySQL8Dialect redis: host: localhost port: 6379 database: 0 timeout: 2000ms lettuce: pool: max-active: 8 max-wait: -1ms max-idle: 8 min-idle: 0 # Redisson配置 redisson: config: | singleServerConfig: address: redis://${spring.redis.host}:${spring.redis.port} database: ${spring.redis.database} connectionMinimumIdleSize: 5 # 应用配置 app: reward: max-retry-count: 3 # 最大重试次数4. 核心流程拆解如何安全发放奖励整个发放流程的设计目标是原子性、幂等性、可追溯。流程如下请求入口与幂等校验 客户端携带userId,taskId, 以及一个全局唯一的bizNo业务流水号可由客户端生成或服务端生成发起请求。防重检查 服务端首先以bizNo为键查询reward_record表。如果记录已存在且状态为成功直接返回“已发放”如果状态为处理中可能正在处理需根据策略等待或返回“处理中”如果记录不存在则进入下一步。获取分布式锁 以userId为粒度加锁防止同一用户并发领取多个任务导致余额更新错乱。这是解决“更新丢失”的关键。事务内核心操作 a.插入奖励记录 向reward_record插入一条状态为“处理中”的记录。利用数据库的唯一索引 (biz_no)确保同一bizNo不会插入两次这是幂等的第二道防线。 b.更新用户余额 使用乐观锁通过version字段更新user_account表增加余额。如果更新行数为0说明并发冲突抛出异常并回滚。 c.插入账户流水 向account_flow插入一条入账流水记录变动前后余额。更新奖励记录状态 在事务提交后或作为事务的最后一步将reward_record的状态更新为“成功”。释放分布式锁。异步补偿与对账可选 对于更复杂的场景如需要调用外部系统可以在事务成功后发送消息到MQ由消费者异步处理并定期对账保证最终一致性。5. 完整代码实现5.1 实体类定义// UserAccount.java package com.example.rewardsystem.entity; import jakarta.persistence.*; import lombok.Data; import org.hibernate.annotations.CreationTimestamp; import org.hibernate.annotations.UpdateTimestamp; import java.math.BigDecimal; import java.time.LocalDateTime; Entity Table(name user_account) Data public class UserAccount { Id GeneratedValue(strategy GenerationType.IDENTITY) private Long id; Column(name user_id, nullable false, unique true, length 64) private String userId; Column(name balance, nullable false, precision 15, scale 2) private BigDecimal balance BigDecimal.ZERO; Version // 乐观锁版本号 Column(name version, nullable false) private Integer version 0; CreationTimestamp Column(name created_time, updatable false) private LocalDateTime createdTime; UpdateTimestamp Column(name updated_time) private LocalDateTime updatedTime; }// RewardRecord.java package com.example.rewardsystem.entity; import jakarta.persistence.*; import lombok.Data; import org.hibernate.annotations.CreationTimestamp; import org.hibernate.annotations.UpdateTimestamp; import java.math.BigDecimal; import java.time.LocalDateTime; Entity Table(name reward_record, indexes { Index(name idx_biz_no, columnList bizNo), Index(name idx_user_task, columnList userId, taskId), Index(name idx_status_created, columnList status, createdTime) }) Data public class RewardRecord { Id GeneratedValue(strategy GenerationType.IDENTITY) private Long id; Column(name biz_no, nullable false, unique true, length 128) private String bizNo; // 幂等关键字段 Column(name user_id, nullable false, length 64) private String userId; Column(name task_id, nullable false) private Long taskId; Column(name reward_amount, nullable false, precision 10, scale 2) private BigDecimal rewardAmount; Column(name status, nullable false) private Integer status; // 0-处理中1-成功2-失败3-已撤销 Column(name retry_count) private Integer retryCount 0; Column(name error_msg, columnDefinition TEXT) private String errorMsg; CreationTimestamp Column(name created_time, updatable false) private LocalDateTime createdTime; UpdateTimestamp Column(name updated_time) private LocalDateTime updatedTime; // 状态枚举 public static class Status { public static final int PROCESSING 0; public static final int SUCCESS 1; public static final int FAILED 2; public static final int REVOKED 3; } }5.2 核心服务层实现// RewardService.java package com.example.rewardsystem.service; import com.example.rewardsystem.entity.*; import com.example.rewardsystem.repository.RewardRecordRepository; import com.example.rewardsystem.repository.UserAccountRepository; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.redisson.api.RLock; import org.redisson.api.RedissonClient; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import java.math.BigDecimal; import java.time.LocalDateTime; import java.util.Optional; import java.util.concurrent.TimeUnit; Service Slf4j RequiredArgsConstructor public class RewardService { private final UserAccountRepository userAccountRepository; private final RewardRecordRepository rewardRecordRepository; private final AccountFlowService accountFlowService; private final RedissonClient redissonClient; /** * 发放奖励 - 核心方法 * param userId 用户ID * param taskId 任务ID * param rewardAmount 奖励金额 * param bizNo 业务流水号 (必须全局唯一) * return 奖励记录ID */ public Long grantReward(String userId, Long taskId, BigDecimal rewardAmount, String bizNo) { // 1. 幂等性检查根据bizNo查询是否已处理 OptionalRewardRecord existingRecord rewardRecordRepository.findByBizNo(bizNo); if (existingRecord.isPresent()) { RewardRecord record existingRecord.get(); if (RewardRecord.Status.SUCCESS record.getStatus()) { log.info(业务流水号已处理成功直接返回。bizNo: {}, recordId: {}, bizNo, record.getId()); return record.getId(); // 幂等返回 } if (RewardRecord.Status.PROCESSING record.getStatus()) { // 处理中可能上游重试这里可以根据业务策略决定等待、返回特定状态或抛异常 log.warn(业务流水号正在处理中请勿重复请求。bizNo: {}, bizNo); throw new RuntimeException(请求正在处理中请稍后查询); } // 如果是失败状态可以考虑重试逻辑这里简单抛异常 log.warn(业务流水号对应记录处于失败状态。bizNo: {}, status: {}, bizNo, record.getStatus()); throw new RuntimeException(之前的请求处理失败请联系管理员); } // 2. 获取用户级别的分布式锁防止同一用户并发操作 String lockKey reward:lock:user: userId; RLock lock redissonClient.getLock(lockKey); boolean locked false; try { // 尝试加锁最多等待3秒锁持有时间10秒应大于事务执行时间 locked lock.tryLock(3, 10, TimeUnit.SECONDS); if (!locked) { log.error(获取用户锁超时userId: {}, userId); throw new RuntimeException(系统繁忙请稍后重试); } // 3. 在锁的保护下执行事务性操作 return doGrantRewardInTransaction(userId, taskId, rewardAmount, bizNo); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException(系统中断异常, e); } finally { if (locked lock.isHeldByCurrentThread()) { lock.unlock(); } } } /** * 事务内执行奖励发放 */ Transactional(rollbackFor Exception.class) protected Long doGrantRewardInTransaction(String userId, Long taskId, BigDecimal rewardAmount, String bizNo) { // 4. 插入处理中记录 (利用唯一索引防并发插入) RewardRecord record new RewardRecord(); record.setBizNo(bizNo); record.setUserId(userId); record.setTaskId(taskId); record.setRewardAmount(rewardAmount); record.setStatus(RewardRecord.Status.PROCESSING); rewardRecordRepository.save(record); // 如果bizNo重复这里会抛出DataIntegrityViolationException // 5. 查询并更新用户账户使用乐观锁 UserAccount account userAccountRepository.findByUserIdForUpdate(userId); // 也可以用悲观锁 select for update if (account null) { // 账户不存在初始化一个根据业务决定 account new UserAccount(); account.setUserId(userId); account.setBalance(BigDecimal.ZERO); account userAccountRepository.save(account); } BigDecimal oldBalance account.getBalance(); BigDecimal newBalance oldBalance.add(rewardAmount); // 乐观锁更新 int updateCount userAccountRepository.updateBalanceWithVersion( userId, newBalance, account.getVersion(), account.getVersion() 1); if (updateCount 0) { // 乐观锁冲突说明在查询和更新之间余额被其他操作修改了 log.error(更新用户余额时发生乐观锁冲突userId: {}, oldVersion: {}, userId, account.getVersion()); throw new RuntimeException(并发操作冲突请重试); } // 6. 插入账户流水 accountFlowService.createFlow( userId, rewardAmount, oldBalance, newBalance, REWARD, bizNo, 任务奖励发放任务ID: taskId ); // 7. 更新奖励记录状态为成功 record.setStatus(RewardRecord.Status.SUCCESS); rewardRecordRepository.save(record); // 或使用 updateStatusById log.info(奖励发放成功。userId: {}, taskId: {}, amount: {}, bizNo: {}, recordId: {}, userId, taskId, rewardAmount, bizNo, record.getId()); return record.getId(); } }// UserAccountRepository.java (片段) package com.example.rewardsystem.repository; import com.example.rewardsystem.entity.UserAccount; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.data.jpa.repository.Modifying; import org.springframework.data.jpa.repository.Query; import org.springframework.data.repository.query.Param; import org.springframework.transaction.annotation.Transactional; import java.util.Optional; public interface UserAccountRepository extends JpaRepositoryUserAccount, Long { OptionalUserAccount findByUserId(String userId); // 使用悲观锁在事务中锁定行 Query(SELECT ua FROM UserAccount ua WHERE ua.userId :userId) UserAccount findByUserIdForUpdate(Param(userId) String userId); // 乐观锁更新 Modifying Transactional Query(UPDATE UserAccount ua SET ua.balance :newBalance, ua.version :newVersion WHERE ua.userId :userId AND ua.version :oldVersion) int updateBalanceWithVersion(Param(userId) String userId, Param(newBalance) BigDecimal newBalance, Param(oldVersion) Integer oldVersion, Param(newVersion) Integer newVersion); }5.3 控制器层与请求对象// RewardRequest.java package com.example.rewardsystem.dto; import jakarta.validation.constraints.DecimalMin; import jakarta.validation.constraints.NotBlank; import jakarta.validation.constraints.NotNull; import lombok.Data; import java.math.BigDecimal; Data public class RewardRequest { NotBlank(message 用户ID不能为空) private String userId; NotNull(message 任务ID不能为空) private Long taskId; NotNull(message 奖励金额不能为空) DecimalMin(value 0.01, message 奖励金额必须大于0) private BigDecimal rewardAmount; NotBlank(message 业务流水号不能为空) private String bizNo; // 建议由客户端生成UUID或由服务端统一生成 }// RewardController.java package com.example.rewardsystem.controller; import com.example.rewardsystem.dto.RewardRequest; import com.example.rewardsystem.service.RewardService; import jakarta.validation.Valid; import lombok.RequiredArgsConstructor; import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.*; import java.util.HashMap; import java.util.Map; RestController RequestMapping(/api/reward) RequiredArgsConstructor public class RewardController { private final RewardService rewardService; PostMapping(/grant) public ResponseEntityMapString, Object grantReward(Valid RequestBody RewardRequest request) { try { Long recordId rewardService.grantReward( request.getUserId(), request.getTaskId(), request.getRewardAmount(), request.getBizNo() ); MapString, Object result new HashMap(); result.put(code, 200); result.put(message, 奖励发放成功); result.put(data, Map.of(recordId, recordId)); return ResponseEntity.ok(result); } catch (RuntimeException e) { // 这里应该根据异常类型细化错误码 MapString, Object result new HashMap(); result.put(code, 500); result.put(message, 奖励发放失败: e.getMessage()); return ResponseEntity.internalServerError().body(result); } } }6. 运行结果与效果验证启动Spring Boot应用后我们可以使用curl或 Postman 进行测试。1. 首次正常请求curl -X POST http://localhost:8080/api/reward/grant \ -H Content-Type: application/json \ -d { userId: user_001, taskId: 1001, rewardAmount: 0.10, bizNo: REWARD_20240520_001 }预期成功响应{ code: 200, message: 奖励发放成功, data: { recordId: 1 } }此时查询数据库reward_record表会有一条biz_no为REWARD_20240520_001状态为1成功的记录。user_account表中user_001的balance会增加 0.10version会增加 1。account_flow表会有一条对应的流水记录。2. 使用相同bizNo重复请求模拟客户端重试再次执行完全相同的curl命令。预期响应幂等生效如果我们的服务是严格按照上述逻辑在grantReward方法开头就根据bizNo查询到了成功记录那么应该快速返回成功并且不会再次增加余额。响应体中的recordId应该是第一次请求生成的ID。检查数据库用户余额不会再次增加流水表不会插入新记录。这就防止了“重复发放”。3. 模拟并发请求使用工具如JMeter使用同一个userId但不同的bizNo在极短时间内发起多个请求。由于我们使用了基于userId的分布式锁这些请求会串行执行从而避免了“更新丢失”保证了余额增加的准确性。如果不用锁很可能出现文章开头说的“收益折半”问题。4. 验证乐观锁可以尝试在updateBalanceWithVersion方法执行前手动修改数据库中的version字段。再次请求时updateCount会为0服务会抛出“并发操作冲突”异常事务回滚保证了数据一致性。7. 常见问题与排查思路问题现象可能原因排查方式解决方案错误Duplicate entry xxx for key reward_record.biz_no业务流水号bizNo重复。可能是客户端重复生成或上游系统重试。1. 检查客户端生成bizNo的规则是否保证全局唯一。2. 查看日志确认是否已存在相同bizNo的成功记录。1. 确保bizNo生成算法包含时间戳、随机数、业务标识等。2. 在服务入口加强幂等检查对于已成功的请求直接返回。错误Lock wait timeout exceeded数据库行锁或表锁等待超时。可能是一个长事务持有了锁或者分布式锁未正确释放。1. 查看数据库的information_schema.INNODB_LOCKS和INNODB_LOCK_WAITS表。2. 检查应用日志看是否有事务执行时间过长。3. 检查Redisson锁的leaseTime是否设置过短。1. 优化事务内的SQL减少锁持有时间。2. 检查是否在事务中进行了远程调用等耗时操作。3. 适当增加分布式锁的leaseTime并确保在finally块中释放锁。用户余额增加了但奖励记录状态还是“处理中”事务提交后更新奖励记录状态的操作可能失败了如网络抖动、唯一键冲突等。1. 查询reward_record表确认记录状态。2. 查看应用错误日志寻找更新失败的异常。1. 将状态更新放在事务内作为最后一步利用事务原子性保证。2. 实现一个状态补偿Job定期扫描“处理中”但已过期的记录根据账户流水进行核对并修复状态。分布式锁未释放导致后续请求一直等待持有锁的线程执行时间超过了锁的leaseTime或者线程异常终止未执行finally块。1. 查看Redis中锁的Key是否还存在。2. 分析应用日志寻找线程中断或超时的记录。1. 合理设置锁的leaseTime确保大于业务最大执行时间。2. 使用lock.tryLock( waitTime, leaseTime, unit)并在try-catch-finally中确保锁被释放。3. 考虑使用Redisson的看门狗机制自动续期。“收益折半”问题依然出现1. 乐观锁更新失败后业务直接返回了错误但没有让客户端重试。2. 并发时多个请求都通过了初始余额查询然后依次更新。1. 检查乐观锁更新返回值是否为0。2. 检查是否在用户粒度上加了分布式锁。1. 在乐观锁冲突时可以设计重试机制如最多重试3次。2.必须在操作同一用户余额时加分布式锁或使用SELECT ... FOR UPDATE进行悲观锁。流水记录与余额对不上1. 插入流水和更新余额不在同一个事务中。2. 异步操作失败导致状态不一致。1. 核对account_flow的balance_before和balance_after与user_account的历史快照。2. 检查是否有未补偿的失败异步任务。1.确保核心的资金操作在同一个数据库事务中。2. 建立每日对账任务核对账户总余额与流水汇总金额。8. 最佳实践与工程建议业务流水号设计不要使用数据库自增ID作为幂等依据因为它只在单库唯一。推荐格式业务类型 日期 随机数/序列号 用户标识/机器标识例如REWARD_20240520_01_ABC123_user001。也可以使用UUID但要注意其无序性可能影响数据库索引性能。锁的粒度与范围锁粒度要细我们锁的是userId而不是整个服务。这提高了并发度。锁范围要精准锁只保护“查询用户账户 - 更新余额”这个关键区间。不要在锁内进行网络IO等耗时操作。设置合理的超时时间分布式锁的waitTime获取锁的等待时间和leaseTime锁的持有时间需要根据业务平均耗时仔细设置。事务边界清晰将数据库更新操作放在一个事务中。远程HTTP调用、发送MQ消息等操作应放在事务提交之后或者通过本地事务消息表等模式保证最终一致性。避免大事务事务内操作要快。状态机驱动对于reward_record的状态处理中、成功、失败、已撤销定义明确的状态转换图。任何状态变更都必须符合前置状态条件。例如只有“成功”状态的记录才能被“撤销”。补偿与对账实现一个后台Job定期扫描状态为“处理中”但创建时间超过阈值的reward_record。根据account_flow和user_account进行核对自动修复数据或发出告警由人工介入。这是保证系统最终一致性的最后一道防线。监控与告警监控关键指标奖励发放成功率、平均耗时、乐观锁冲突次数、分布式锁获取失败率。对失败率升高、大量“处理中”记录堆积等情况设置告警。测试策略单元测试覆盖正常流程、幂等、乐观锁冲突等场景。集成测试启动完整的Spring上下文测试数据库和Redis的交互。压力测试使用JMeter模拟高并发领取验证余额准确性。通过以上从原理分析、环境搭建、代码实现到问题排查的完整闭环我们构建了一个能够有效抵御“收益折半”、“状态未退”等数据一致性问题的奖励发放系统。这套方案的核心思想——幂等防重、锁防并发、事务保原子、对账保最终——可以广泛应用于任何需要保证数据准确性的业务场景如支付、订单、库存、积分等。技术的价值在于解决真实世界的问题。下次再遇到看似“灵异”的Bug时不妨先从这些基础且强大的机制入手排查你会发现大多数“恶作剧”背后都有一套严谨的逻辑等待你去发现和构建。
Spring Boot实战：基于幂等性与分布式锁解决高并发奖励发放数据一致性问题
最近在技术社区看到一个很有意思的现象一个看似“无厘头”的标题背后却可能隐藏着开发者对数据一致性、状态管理和异常处理的深刻困惑。标题里提到的“未退”、“收益折半”、“突然躺中”像极了我们在处理异步任务、金融计算或分布式事务时因为一个状态没更新、一个锁没释放或者一次幂等性没做好导致的数据“灵异事件”。这绝不是小孩的恶作剧而是系统在“恶作剧”开发者。今天我们就来彻底拆解这类问题。我将从一个高并发场景下的“收益计算异常”案例入手带你理解事务、幂等性和最终一致性的核心概念并用一套可落地的Spring Boot Redis MySQL方案演示如何从代码层面杜绝“收益折半”、“状态未退”的坑。如果你正在开发涉及支付、积分、任务奖励等有状态变化的系统这篇文章值得你仔细读完。1. 从“收益折半”案例看状态同步的核心痛点我们先还原一下标题中可能描述的场景一个用户“宝宝”完成了一个动作比如观看直播、完成任务预期获得0.1的收益“软米”。但系统实际只发放了0.05收益“折半”。同时用户可能发起了“退款”或“撤销”操作但系统状态显示“未退”仿佛被“涮”了。在技术层面这通常指向几个经典问题非原子性操作 “收益发放”不是一个原子操作。它可能被拆分为“计算收益”、“更新账户余额”、“记录流水”等多个步骤。如果系统在中间步骤崩溃或并发冲突就会导致状态不一致如流水记了余额没加。并发更新丢失 高并发下两个线程同时读取用户余额假设为0都加上0.1然后先后写回结果余额变成了0.1而不是0.2。这就是“更新丢失”感觉收益被“吞”了一半。缺乏幂等性 用户请求可能因为网络超时被客户端重试如果服务端没有识别出这是同一个请求就会重复处理导致收益被发放两次或者“退款”操作执行两次。但有时因为并发和状态判断错误可能只成功了一次看起来像“折半”或“未退”。状态机设计混乱 “已发放”、“发放中”、“已退款”、“退款中”这些状态如果没有清晰的定义和严格的转换规则就容易出现“未退”但钱已扣或者“已退”但状态没更新的情况。本文的核心判断是大多数这类“灵异”问题根源不在于复杂的算法而在于对“事务边界”、“并发控制”和“状态管理”这些基础机制的理解不到位或实现有瑕疵。接下来我们将通过一个模拟的“任务奖励发放系统”从原理到实战构建一个健壮的解决方案。2. 核心概念事务、幂等性与最终一致性在深入代码之前必须厘清三个基石概念。2.1 数据库事务 (ACID)事务是保证单个数据库内数据操作“要么全做要么全不做”的机制。其ACID特性原子性 (Atomicity) 事务内的操作是一个不可分割的整体。这正是解决“收益折半”的第一步。如果发放收益涉及更新用户表和插入流水表它们必须在同一个事务中。一致性 (Consistency) 事务执行前后数据库必须保持一致性状态如余额不能为负。隔离性 (Isolation) 并发事务之间相互隔离防止脏读、不可重复读、幻读。隔离级别如读已提交、可重复读的选择直接影响并发问题。持久性 (Durability) 事务提交后对数据的修改是永久性的。在分布式环境下单纯依赖数据库事务往往不够因为我们的服务、缓存、消息队列可能不在同一个数据库实例上。2.2 幂等性 (Idempotence)这是一个数学概念在计算机科学中指无论操作执行一次还是多次其产生的结果都是相同的。对于HTTP接口意味着客户端用同样的参数重复调用服务端的状态变化只发生一次。为什么需要幂等网络是不稳定的。用户点击“领取奖励”按钮可能因为响应慢而多次点击客户端超时后自动重试消息队列消费者失败后重新投递。如果没有幂等控制就会导致重复发放。实现幂等性的常见手段 唯一业务流水号、数据库唯一索引、Token机制、状态机判断。2.3 最终一致性 (Eventual Consistency)这是分布式系统领域CAP理论中的一种一致性模型。它不要求数据时刻保持强一致但保证在没有新的更新操作后经过一段时间所有副本的数据最终会达到一致状态。与我们场景的关联 当“发放收益”需要调用外部积分系统、会计系统时我们可能采用“先更新本地数据库状态为‘发放中’然后异步调用外部系统成功后更新为‘已发放’”的模式。这期间用户查询到的状态可能是“发放中”但外部系统可能已处理完成这就是短暂的“不一致”。我们需要通过补偿机制如对账、重试来达到最终一致。3. 环境准备与项目初始化我们将创建一个Spring Boot项目来模拟奖励发放系统。技术栈JDK: 17 或 21Spring Boot: 3.1.x 或 3.2.x持久层: Spring Data JPA MySQL 8.0缓存与分布式锁: Spring Data Redis Redisson构建工具: Maven数据库表设计user_account用户账户表存储余额。reward_task奖励任务表定义任务和奖励金额。reward_record奖励发放记录表核心表用于实现幂等和状态追踪。account_flow账户流水表记录每一笔变动。使用schema.sql初始化数据库-- schema.sql CREATE DATABASE IF NOT EXISTS reward_system DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; USE reward_system; -- 用户账户表 CREATE TABLE user_account ( id BIGINT PRIMARY KEY AUTO_INCREMENT COMMENT 主键, user_id VARCHAR(64) NOT NULL UNIQUE COMMENT 用户唯一标识, balance DECIMAL(15, 2) NOT NULL DEFAULT 0.00 COMMENT 账户余额, version INT NOT NULL DEFAULT 0 COMMENT 乐观锁版本号, created_time DATETIME DEFAULT CURRENT_TIMESTAMP, updated_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_user_id (user_id) ) COMMENT 用户账户; -- 奖励任务表 CREATE TABLE reward_task ( id BIGINT PRIMARY KEY AUTO_INCREMENT, task_code VARCHAR(64) NOT NULL UNIQUE COMMENT 任务编码, task_name VARCHAR(255) NOT NULL COMMENT 任务名称, reward_amount DECIMAL(10, 2) NOT NULL COMMENT 奖励金额, is_active BOOLEAN DEFAULT TRUE COMMENT 是否生效, created_time DATETIME DEFAULT CURRENT_TIMESTAMP ) COMMENT 奖励任务定义; -- 奖励发放记录表 (幂等性核心) CREATE TABLE reward_record ( id BIGINT PRIMARY KEY AUTO_INCREMENT, biz_no VARCHAR(128) NOT NULL UNIQUE COMMENT 业务唯一流水号(用于幂等), user_id VARCHAR(64) NOT NULL COMMENT 用户ID, task_id BIGINT NOT NULL COMMENT 任务ID, reward_amount DECIMAL(10, 2) NOT NULL COMMENT 奖励金额, status TINYINT NOT NULL COMMENT 状态: 0-处理中, 1-发放成功, 2-发放失败, 3-已撤销, retry_count INT DEFAULT 0 COMMENT 重试次数, error_msg TEXT COMMENT 错误信息, created_time DATETIME DEFAULT CURRENT_TIMESTAMP, updated_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_biz_no (biz_no), INDEX idx_user_task (user_id, task_id), INDEX idx_status_created (status, created_time) ) COMMENT 奖励发放记录; -- 账户流水表 CREATE TABLE account_flow ( id BIGINT PRIMARY KEY AUTO_INCREMENT, flow_no VARCHAR(128) NOT NULL UNIQUE COMMENT 流水号, user_id VARCHAR(64) NOT NULL COMMENT 用户ID, amount DECIMAL(10, 2) NOT NULL COMMENT 变动金额正为入账负为出账, balance_before DECIMAL(15, 2) NOT NULL COMMENT 变动前余额, balance_after DECIMAL(15, 2) NOT NULL COMMENT 变动后余额, biz_type VARCHAR(32) NOT NULL COMMENT 业务类型: REWARD-任务奖励, REFUND-退款, biz_no VARCHAR(128) NOT NULL COMMENT 关联业务号(如reward_record.biz_no), remark VARCHAR(512) COMMENT 备注, created_time DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_user_id (user_id), INDEX idx_biz_no (biz_no), INDEX idx_created_time (created_time) ) COMMENT 账户流水;Maven依赖 (pom.xml):?xml version1.0 encodingUTF-8? project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instance xsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd modelVersion4.0.0/modelVersion parent groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-parent/artifactId version3.1.6/version !-- 请根据实际情况调整 -- relativePath/ /parent groupIdcom.example/groupId artifactIdreward-system/artifactId version0.0.1-SNAPSHOT/version namereward-system/name descriptionDemo project for reward distribution/description properties java.version17/java.version /properties dependencies !-- Web -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency !-- JPA -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-data-jpa/artifactId /dependency !-- MySQL Driver -- dependency groupIdcom.mysql/groupId artifactIdmysql-connector-j/artifactId scoperuntime/scope /dependency !-- Redis -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-data-redis/artifactId /dependency !-- Redisson for distributed lock -- dependency groupIdorg.redisson/groupId artifactIdredisson-spring-boot-starter/artifactId version3.23.5/version !-- 请检查最新版本 -- /dependency !-- Lombok -- dependency groupIdorg.projectlombok/groupId artifactIdlombok/artifactId optionaltrue/optional /dependency !-- Test -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-test/artifactId scopetest/scope /dependency /dependencies build plugins plugin groupIdorg.springframework.boot/groupId artifactIdspring-boot-maven-plugin/artifactId configuration excludes exclude groupIdorg.projectlombok/groupId artifactIdlombok/artifactId /exclude /excludes /configuration /plugin /plugins /build /project应用配置文件 (application.yml):# application.yml spring: datasource: url: jdbc:mysql://localhost:3306/reward_system?useUnicodetruecharacterEncodingutf8useSSLfalseserverTimezoneAsia/Shanghai username: root password: your_password driver-class-name: com.mysql.cj.jdbc.Driver jpa: hibernate: ddl-auto: update # 首次启动可设为update创建表生产环境建议使用none通过sql脚本管理 show-sql: true properties: hibernate: format_sql: true dialect: org.hibernate.dialect.MySQL8Dialect redis: host: localhost port: 6379 database: 0 timeout: 2000ms lettuce: pool: max-active: 8 max-wait: -1ms max-idle: 8 min-idle: 0 # Redisson配置 redisson: config: | singleServerConfig: address: redis://${spring.redis.host}:${spring.redis.port} database: ${spring.redis.database} connectionMinimumIdleSize: 5 # 应用配置 app: reward: max-retry-count: 3 # 最大重试次数4. 核心流程拆解如何安全发放奖励整个发放流程的设计目标是原子性、幂等性、可追溯。流程如下请求入口与幂等校验 客户端携带userId,taskId, 以及一个全局唯一的bizNo业务流水号可由客户端生成或服务端生成发起请求。防重检查 服务端首先以bizNo为键查询reward_record表。如果记录已存在且状态为成功直接返回“已发放”如果状态为处理中可能正在处理需根据策略等待或返回“处理中”如果记录不存在则进入下一步。获取分布式锁 以userId为粒度加锁防止同一用户并发领取多个任务导致余额更新错乱。这是解决“更新丢失”的关键。事务内核心操作 a.插入奖励记录 向reward_record插入一条状态为“处理中”的记录。利用数据库的唯一索引 (biz_no)确保同一bizNo不会插入两次这是幂等的第二道防线。 b.更新用户余额 使用乐观锁通过version字段更新user_account表增加余额。如果更新行数为0说明并发冲突抛出异常并回滚。 c.插入账户流水 向account_flow插入一条入账流水记录变动前后余额。更新奖励记录状态 在事务提交后或作为事务的最后一步将reward_record的状态更新为“成功”。释放分布式锁。异步补偿与对账可选 对于更复杂的场景如需要调用外部系统可以在事务成功后发送消息到MQ由消费者异步处理并定期对账保证最终一致性。5. 完整代码实现5.1 实体类定义// UserAccount.java package com.example.rewardsystem.entity; import jakarta.persistence.*; import lombok.Data; import org.hibernate.annotations.CreationTimestamp; import org.hibernate.annotations.UpdateTimestamp; import java.math.BigDecimal; import java.time.LocalDateTime; Entity Table(name user_account) Data public class UserAccount { Id GeneratedValue(strategy GenerationType.IDENTITY) private Long id; Column(name user_id, nullable false, unique true, length 64) private String userId; Column(name balance, nullable false, precision 15, scale 2) private BigDecimal balance BigDecimal.ZERO; Version // 乐观锁版本号 Column(name version, nullable false) private Integer version 0; CreationTimestamp Column(name created_time, updatable false) private LocalDateTime createdTime; UpdateTimestamp Column(name updated_time) private LocalDateTime updatedTime; }// RewardRecord.java package com.example.rewardsystem.entity; import jakarta.persistence.*; import lombok.Data; import org.hibernate.annotations.CreationTimestamp; import org.hibernate.annotations.UpdateTimestamp; import java.math.BigDecimal; import java.time.LocalDateTime; Entity Table(name reward_record, indexes { Index(name idx_biz_no, columnList bizNo), Index(name idx_user_task, columnList userId, taskId), Index(name idx_status_created, columnList status, createdTime) }) Data public class RewardRecord { Id GeneratedValue(strategy GenerationType.IDENTITY) private Long id; Column(name biz_no, nullable false, unique true, length 128) private String bizNo; // 幂等关键字段 Column(name user_id, nullable false, length 64) private String userId; Column(name task_id, nullable false) private Long taskId; Column(name reward_amount, nullable false, precision 10, scale 2) private BigDecimal rewardAmount; Column(name status, nullable false) private Integer status; // 0-处理中1-成功2-失败3-已撤销 Column(name retry_count) private Integer retryCount 0; Column(name error_msg, columnDefinition TEXT) private String errorMsg; CreationTimestamp Column(name created_time, updatable false) private LocalDateTime createdTime; UpdateTimestamp Column(name updated_time) private LocalDateTime updatedTime; // 状态枚举 public static class Status { public static final int PROCESSING 0; public static final int SUCCESS 1; public static final int FAILED 2; public static final int REVOKED 3; } }5.2 核心服务层实现// RewardService.java package com.example.rewardsystem.service; import com.example.rewardsystem.entity.*; import com.example.rewardsystem.repository.RewardRecordRepository; import com.example.rewardsystem.repository.UserAccountRepository; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.redisson.api.RLock; import org.redisson.api.RedissonClient; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import java.math.BigDecimal; import java.time.LocalDateTime; import java.util.Optional; import java.util.concurrent.TimeUnit; Service Slf4j RequiredArgsConstructor public class RewardService { private final UserAccountRepository userAccountRepository; private final RewardRecordRepository rewardRecordRepository; private final AccountFlowService accountFlowService; private final RedissonClient redissonClient; /** * 发放奖励 - 核心方法 * param userId 用户ID * param taskId 任务ID * param rewardAmount 奖励金额 * param bizNo 业务流水号 (必须全局唯一) * return 奖励记录ID */ public Long grantReward(String userId, Long taskId, BigDecimal rewardAmount, String bizNo) { // 1. 幂等性检查根据bizNo查询是否已处理 OptionalRewardRecord existingRecord rewardRecordRepository.findByBizNo(bizNo); if (existingRecord.isPresent()) { RewardRecord record existingRecord.get(); if (RewardRecord.Status.SUCCESS record.getStatus()) { log.info(业务流水号已处理成功直接返回。bizNo: {}, recordId: {}, bizNo, record.getId()); return record.getId(); // 幂等返回 } if (RewardRecord.Status.PROCESSING record.getStatus()) { // 处理中可能上游重试这里可以根据业务策略决定等待、返回特定状态或抛异常 log.warn(业务流水号正在处理中请勿重复请求。bizNo: {}, bizNo); throw new RuntimeException(请求正在处理中请稍后查询); } // 如果是失败状态可以考虑重试逻辑这里简单抛异常 log.warn(业务流水号对应记录处于失败状态。bizNo: {}, status: {}, bizNo, record.getStatus()); throw new RuntimeException(之前的请求处理失败请联系管理员); } // 2. 获取用户级别的分布式锁防止同一用户并发操作 String lockKey reward:lock:user: userId; RLock lock redissonClient.getLock(lockKey); boolean locked false; try { // 尝试加锁最多等待3秒锁持有时间10秒应大于事务执行时间 locked lock.tryLock(3, 10, TimeUnit.SECONDS); if (!locked) { log.error(获取用户锁超时userId: {}, userId); throw new RuntimeException(系统繁忙请稍后重试); } // 3. 在锁的保护下执行事务性操作 return doGrantRewardInTransaction(userId, taskId, rewardAmount, bizNo); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException(系统中断异常, e); } finally { if (locked lock.isHeldByCurrentThread()) { lock.unlock(); } } } /** * 事务内执行奖励发放 */ Transactional(rollbackFor Exception.class) protected Long doGrantRewardInTransaction(String userId, Long taskId, BigDecimal rewardAmount, String bizNo) { // 4. 插入处理中记录 (利用唯一索引防并发插入) RewardRecord record new RewardRecord(); record.setBizNo(bizNo); record.setUserId(userId); record.setTaskId(taskId); record.setRewardAmount(rewardAmount); record.setStatus(RewardRecord.Status.PROCESSING); rewardRecordRepository.save(record); // 如果bizNo重复这里会抛出DataIntegrityViolationException // 5. 查询并更新用户账户使用乐观锁 UserAccount account userAccountRepository.findByUserIdForUpdate(userId); // 也可以用悲观锁 select for update if (account null) { // 账户不存在初始化一个根据业务决定 account new UserAccount(); account.setUserId(userId); account.setBalance(BigDecimal.ZERO); account userAccountRepository.save(account); } BigDecimal oldBalance account.getBalance(); BigDecimal newBalance oldBalance.add(rewardAmount); // 乐观锁更新 int updateCount userAccountRepository.updateBalanceWithVersion( userId, newBalance, account.getVersion(), account.getVersion() 1); if (updateCount 0) { // 乐观锁冲突说明在查询和更新之间余额被其他操作修改了 log.error(更新用户余额时发生乐观锁冲突userId: {}, oldVersion: {}, userId, account.getVersion()); throw new RuntimeException(并发操作冲突请重试); } // 6. 插入账户流水 accountFlowService.createFlow( userId, rewardAmount, oldBalance, newBalance, REWARD, bizNo, 任务奖励发放任务ID: taskId ); // 7. 更新奖励记录状态为成功 record.setStatus(RewardRecord.Status.SUCCESS); rewardRecordRepository.save(record); // 或使用 updateStatusById log.info(奖励发放成功。userId: {}, taskId: {}, amount: {}, bizNo: {}, recordId: {}, userId, taskId, rewardAmount, bizNo, record.getId()); return record.getId(); } }// UserAccountRepository.java (片段) package com.example.rewardsystem.repository; import com.example.rewardsystem.entity.UserAccount; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.data.jpa.repository.Modifying; import org.springframework.data.jpa.repository.Query; import org.springframework.data.repository.query.Param; import org.springframework.transaction.annotation.Transactional; import java.util.Optional; public interface UserAccountRepository extends JpaRepositoryUserAccount, Long { OptionalUserAccount findByUserId(String userId); // 使用悲观锁在事务中锁定行 Query(SELECT ua FROM UserAccount ua WHERE ua.userId :userId) UserAccount findByUserIdForUpdate(Param(userId) String userId); // 乐观锁更新 Modifying Transactional Query(UPDATE UserAccount ua SET ua.balance :newBalance, ua.version :newVersion WHERE ua.userId :userId AND ua.version :oldVersion) int updateBalanceWithVersion(Param(userId) String userId, Param(newBalance) BigDecimal newBalance, Param(oldVersion) Integer oldVersion, Param(newVersion) Integer newVersion); }5.3 控制器层与请求对象// RewardRequest.java package com.example.rewardsystem.dto; import jakarta.validation.constraints.DecimalMin; import jakarta.validation.constraints.NotBlank; import jakarta.validation.constraints.NotNull; import lombok.Data; import java.math.BigDecimal; Data public class RewardRequest { NotBlank(message 用户ID不能为空) private String userId; NotNull(message 任务ID不能为空) private Long taskId; NotNull(message 奖励金额不能为空) DecimalMin(value 0.01, message 奖励金额必须大于0) private BigDecimal rewardAmount; NotBlank(message 业务流水号不能为空) private String bizNo; // 建议由客户端生成UUID或由服务端统一生成 }// RewardController.java package com.example.rewardsystem.controller; import com.example.rewardsystem.dto.RewardRequest; import com.example.rewardsystem.service.RewardService; import jakarta.validation.Valid; import lombok.RequiredArgsConstructor; import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.*; import java.util.HashMap; import java.util.Map; RestController RequestMapping(/api/reward) RequiredArgsConstructor public class RewardController { private final RewardService rewardService; PostMapping(/grant) public ResponseEntityMapString, Object grantReward(Valid RequestBody RewardRequest request) { try { Long recordId rewardService.grantReward( request.getUserId(), request.getTaskId(), request.getRewardAmount(), request.getBizNo() ); MapString, Object result new HashMap(); result.put(code, 200); result.put(message, 奖励发放成功); result.put(data, Map.of(recordId, recordId)); return ResponseEntity.ok(result); } catch (RuntimeException e) { // 这里应该根据异常类型细化错误码 MapString, Object result new HashMap(); result.put(code, 500); result.put(message, 奖励发放失败: e.getMessage()); return ResponseEntity.internalServerError().body(result); } } }6. 运行结果与效果验证启动Spring Boot应用后我们可以使用curl或 Postman 进行测试。1. 首次正常请求curl -X POST http://localhost:8080/api/reward/grant \ -H Content-Type: application/json \ -d { userId: user_001, taskId: 1001, rewardAmount: 0.10, bizNo: REWARD_20240520_001 }预期成功响应{ code: 200, message: 奖励发放成功, data: { recordId: 1 } }此时查询数据库reward_record表会有一条biz_no为REWARD_20240520_001状态为1成功的记录。user_account表中user_001的balance会增加 0.10version会增加 1。account_flow表会有一条对应的流水记录。2. 使用相同bizNo重复请求模拟客户端重试再次执行完全相同的curl命令。预期响应幂等生效如果我们的服务是严格按照上述逻辑在grantReward方法开头就根据bizNo查询到了成功记录那么应该快速返回成功并且不会再次增加余额。响应体中的recordId应该是第一次请求生成的ID。检查数据库用户余额不会再次增加流水表不会插入新记录。这就防止了“重复发放”。3. 模拟并发请求使用工具如JMeter使用同一个userId但不同的bizNo在极短时间内发起多个请求。由于我们使用了基于userId的分布式锁这些请求会串行执行从而避免了“更新丢失”保证了余额增加的准确性。如果不用锁很可能出现文章开头说的“收益折半”问题。4. 验证乐观锁可以尝试在updateBalanceWithVersion方法执行前手动修改数据库中的version字段。再次请求时updateCount会为0服务会抛出“并发操作冲突”异常事务回滚保证了数据一致性。7. 常见问题与排查思路问题现象可能原因排查方式解决方案错误Duplicate entry xxx for key reward_record.biz_no业务流水号bizNo重复。可能是客户端重复生成或上游系统重试。1. 检查客户端生成bizNo的规则是否保证全局唯一。2. 查看日志确认是否已存在相同bizNo的成功记录。1. 确保bizNo生成算法包含时间戳、随机数、业务标识等。2. 在服务入口加强幂等检查对于已成功的请求直接返回。错误Lock wait timeout exceeded数据库行锁或表锁等待超时。可能是一个长事务持有了锁或者分布式锁未正确释放。1. 查看数据库的information_schema.INNODB_LOCKS和INNODB_LOCK_WAITS表。2. 检查应用日志看是否有事务执行时间过长。3. 检查Redisson锁的leaseTime是否设置过短。1. 优化事务内的SQL减少锁持有时间。2. 检查是否在事务中进行了远程调用等耗时操作。3. 适当增加分布式锁的leaseTime并确保在finally块中释放锁。用户余额增加了但奖励记录状态还是“处理中”事务提交后更新奖励记录状态的操作可能失败了如网络抖动、唯一键冲突等。1. 查询reward_record表确认记录状态。2. 查看应用错误日志寻找更新失败的异常。1. 将状态更新放在事务内作为最后一步利用事务原子性保证。2. 实现一个状态补偿Job定期扫描“处理中”但已过期的记录根据账户流水进行核对并修复状态。分布式锁未释放导致后续请求一直等待持有锁的线程执行时间超过了锁的leaseTime或者线程异常终止未执行finally块。1. 查看Redis中锁的Key是否还存在。2. 分析应用日志寻找线程中断或超时的记录。1. 合理设置锁的leaseTime确保大于业务最大执行时间。2. 使用lock.tryLock( waitTime, leaseTime, unit)并在try-catch-finally中确保锁被释放。3. 考虑使用Redisson的看门狗机制自动续期。“收益折半”问题依然出现1. 乐观锁更新失败后业务直接返回了错误但没有让客户端重试。2. 并发时多个请求都通过了初始余额查询然后依次更新。1. 检查乐观锁更新返回值是否为0。2. 检查是否在用户粒度上加了分布式锁。1. 在乐观锁冲突时可以设计重试机制如最多重试3次。2.必须在操作同一用户余额时加分布式锁或使用SELECT ... FOR UPDATE进行悲观锁。流水记录与余额对不上1. 插入流水和更新余额不在同一个事务中。2. 异步操作失败导致状态不一致。1. 核对account_flow的balance_before和balance_after与user_account的历史快照。2. 检查是否有未补偿的失败异步任务。1.确保核心的资金操作在同一个数据库事务中。2. 建立每日对账任务核对账户总余额与流水汇总金额。8. 最佳实践与工程建议业务流水号设计不要使用数据库自增ID作为幂等依据因为它只在单库唯一。推荐格式业务类型 日期 随机数/序列号 用户标识/机器标识例如REWARD_20240520_01_ABC123_user001。也可以使用UUID但要注意其无序性可能影响数据库索引性能。锁的粒度与范围锁粒度要细我们锁的是userId而不是整个服务。这提高了并发度。锁范围要精准锁只保护“查询用户账户 - 更新余额”这个关键区间。不要在锁内进行网络IO等耗时操作。设置合理的超时时间分布式锁的waitTime获取锁的等待时间和leaseTime锁的持有时间需要根据业务平均耗时仔细设置。事务边界清晰将数据库更新操作放在一个事务中。远程HTTP调用、发送MQ消息等操作应放在事务提交之后或者通过本地事务消息表等模式保证最终一致性。避免大事务事务内操作要快。状态机驱动对于reward_record的状态处理中、成功、失败、已撤销定义明确的状态转换图。任何状态变更都必须符合前置状态条件。例如只有“成功”状态的记录才能被“撤销”。补偿与对账实现一个后台Job定期扫描状态为“处理中”但创建时间超过阈值的reward_record。根据account_flow和user_account进行核对自动修复数据或发出告警由人工介入。这是保证系统最终一致性的最后一道防线。监控与告警监控关键指标奖励发放成功率、平均耗时、乐观锁冲突次数、分布式锁获取失败率。对失败率升高、大量“处理中”记录堆积等情况设置告警。测试策略单元测试覆盖正常流程、幂等、乐观锁冲突等场景。集成测试启动完整的Spring上下文测试数据库和Redis的交互。压力测试使用JMeter模拟高并发领取验证余额准确性。通过以上从原理分析、环境搭建、代码实现到问题排查的完整闭环我们构建了一个能够有效抵御“收益折半”、“状态未退”等数据一致性问题的奖励发放系统。这套方案的核心思想——幂等防重、锁防并发、事务保原子、对账保最终——可以广泛应用于任何需要保证数据准确性的业务场景如支付、订单、库存、积分等。技术的价值在于解决真实世界的问题。下次再遇到看似“灵异”的Bug时不妨先从这些基础且强大的机制入手排查你会发现大多数“恶作剧”背后都有一套严谨的逻辑等待你去发现和构建。
