幂等性处理解决方案实战示例

幂等性是指对同一个操作执行一次或多次，产生的结果是相同的。在分布式系统、网络请求和金融交易等场景中，幂等性设计至关重要。下面我将介绍几种常见的幂等性处理方案及其实战示例。

1. 唯一标识符方案

原理：为每个操作分配唯一ID，服务端通过检查该ID是否已处理过来保证幂等性。

实现示例：

// 数据库表设计
CREATE TABLE idempotent_requests (id VARCHAR(64) PRIMARY KEY,user_id BIGINT NOT NULL,operation_type VARCHAR(32) NOT NULL,status VARCHAR(16) NOT NULL,result TEXT,created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);// Java实现示例
@RestController
@RequestMapping("/orders")
public class OrderController {@Autowiredprivate IdempotentRequestRepository idempotentRepo;@PostMappingpublic ResponseEntity<?> createOrder(@RequestHeader("X-Idempotency-Key") String idempotencyKey,@RequestBody OrderRequest request) {// 检查是否已处理过该请求Optional<IdempotentRequest> existing = idempotentRepo.findById(idempotencyKey);if (existing.isPresent()) {// 返回之前的结果return ResponseEntity.ok(existing.get().getResult());}// 处理业务逻辑Order order = orderService.createOrder(request);// 保存幂等记录IdempotentRequest idempotentRequest = new IdempotentRequest();idempotentRequest.setId(idempotencyKey);idempotentRequest.setOperationType("CREATE_ORDER");idempotentRequest.setStatus("COMPLETED");idempotentRequest.setResult(order.toJson());idempotentRepo.save(idempotentRequest);return ResponseEntity.ok(order);}
}

2. 乐观锁方案

原理：通过版本号或条件判断实现更新操作的幂等性。

实现示例：

// 数据库表设计
CREATE TABLE accounts (id BIGINT PRIMARY KEY,user_id BIGINT NOT NULL,balance DECIMAL(19,4) NOT NULL,version INT NOT NULL DEFAULT 0
);// Java实现示例
@Service
public class AccountService {@Autowiredprivate AccountRepository accountRepo;@Transactionalpublic void transfer(Long fromAccountId, Long toAccountId, BigDecimal amount) {Account fromAccount = accountRepo.findById(fromAccountId).orElseThrow(() -> new AccountNotFoundException(fromAccountId));Account toAccount = accountRepo.findById(toAccountId).orElseThrow(() -> new AccountNotFoundException(toAccountId));if (fromAccount.getBalance().compareTo(amount) < 0) {throw new InsufficientBalanceException();}// 使用版本号实现乐观锁int updatedRows = accountRepo.decreaseBalance(fromAccountId, amount, fromAccount.getVersion());if (updatedRows == 0) {throw new OptimisticLockingFailureException("Account update failed");}updatedRows = accountRepo.increaseBalance(toAccountId, amount, toAccount.getVersion());if (updatedRows == 0) {throw new OptimisticLockingFailureException("Account update failed");}}
}

3. 状态机方案

原理：通过定义业务状态流转规则，确保只有符合条件的状态才能进行转换。

实现示例：

// 订单状态枚举
public enum OrderStatus {CREATED, PAID, SHIPPED, DELIVERED, CANCELLED
}// 订单服务
@Service
public class OrderService {@Autowiredprivate OrderRepository orderRepo;@Transactionalpublic void cancelOrder(Long orderId) {Order order = orderRepo.findById(orderId).orElseThrow(() -> new OrderNotFoundException(orderId));// 状态检查确保幂等性if (order.getStatus() == OrderStatus.CANCELLED) {return; // 已经取消，直接返回}// 只有特定状态可以取消if (order.getStatus() != OrderStatus.CREATED && order.getStatus() != OrderStatus.PAID) {throw new IllegalStateException("Order cannot be cancelled in current state");}order.setStatus(OrderStatus.CANCELLED);orderRepo.save(order);// 执行取消后的业务逻辑...}
}

4. 去重表方案

原理：使用单独的表记录已处理的操作，防止重复处理。

实现示例：

// 数据库表设计
CREATE TABLE payment_records (id BIGINT PRIMARY KEY AUTO_INCREMENT,order_id BIGINT NOT NULL,payment_id VARCHAR(64) NOT NULL,amount DECIMAL(19,4) NOT NULL,status VARCHAR(16) NOT NULL,created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,UNIQUE KEY uk_order_payment (order_id, payment_id)
);// Java实现示例
@Service
public class PaymentService {@Autowiredprivate PaymentRecordRepository paymentRecordRepo;@Transactionalpublic PaymentResult processPayment(PaymentRequest request) {// 检查是否已处理过该支付Optional<PaymentRecord> existing = paymentRecordRepo.findByOrderIdAndPaymentId(request.getOrderId(), request.getPaymentId());if (existing.isPresent()) {return existing.get().toResult();}// 处理支付逻辑PaymentResult result = paymentGateway.charge(request);// 保存支付记录PaymentRecord record = new PaymentRecord();record.setOrderId(request.getOrderId());record.setPaymentId(request.getPaymentId());record.setAmount(request.getAmount());record.setStatus(result.getStatus());paymentRecordRepo.save(record);return result;}
}

5. Token机制方案

原理：客户端先获取令牌，服务端验证令牌有效性后处理请求。

实现示例：

// Java实现示例
@RestController
@RequestMapping("/api")
public class ApiController {@Autowiredprivate TokenService tokenService;// 获取令牌@GetMapping("/token")public ResponseEntity<TokenResponse> getToken() {String token = tokenService.generateToken();return ResponseEntity.ok(new TokenResponse(token));}// 提交订单（需要验证令牌）@PostMapping("/order")public ResponseEntity<Order> createOrder(@RequestParam("token") String token,@RequestBody OrderRequest request) {// 验证并消费令牌if (!tokenService.consumeToken(token)) {throw new InvalidTokenException();}// 处理业务逻辑Order order = orderService.createOrder(request);return ResponseEntity.ok(order);}
}@Service
public class TokenService {@Autowiredprivate RedisTemplate<String, String> redisTemplate;// 令牌有效期private static final long TOKEN_EXPIRE_SECONDS = 300;public String generateToken() {String token = UUID.randomUUID().toString();redisTemplate.opsForValue().set("token:" + token, "1", TOKEN_EXPIRE_SECONDS, TimeUnit.SECONDS);return token;}public boolean consumeToken(String token) {String key = "token:" + token;// 使用原子操作删除令牌，防止并发问题Long deleted = redisTemplate.delete(key);return deleted != null && deleted > 0;}
}

6. 消息队列幂等消费

原理：通过消息ID或业务唯一标识实现消息的幂等消费。

RocketMQ实现示例：

// Java消费者示例
@Component
@RocketMQMessageListener(topic = "ORDER_PAYMENT_TOPIC",consumerGroup = "ORDER_PAYMENT_GROUP")
public class OrderPaymentConsumer implements RocketMQListener<MessageExt> {@Autowiredprivate PaymentService paymentService;@Autowiredprivate ConsumedMessageRepository consumedMessageRepo;@Overridepublic void onMessage(MessageExt message) {String messageId = message.getMsgId();String orderId = new String(message.getBody());// 检查是否已处理过该消息if (consumedMessageRepo.existsByMessageId(messageId)) {return; // 已处理，直接返回}try {// 处理支付逻辑paymentService.processPayment(orderId);// 记录已消费消息ConsumedMessage record = new ConsumedMessage();record.setMessageId(messageId);record.setOrderId(orderId);record.setConsumedTime(new Date());consumedMessageRepo.save(record);} catch (Exception e) {// 处理异常，可能需要重试throw new RuntimeException(e);}}
}