分布式并发场景的核心问题与解决方案

文章目录

分布式并发场景的核心问题与解决方案
- 一、核心问题分析
- - 1. 分布式事务问题
  - 2. 数据一致性问题
  - 3. 并发控制问题
  - 4. 分布式锁失效问题
- 二、解决方案
- - 1. 分布式事务解决方案
  - - 1.1 可靠消息最终一致性方案
    - 1.2 TCC方案实现
  - 2. 缓存一致性解决方案
  - - 2.1 延迟双删策略
    - 2.2 Canal方案
  - 3. 并发控制解决方案
  - - 3.1 基于Redis的原子操作
    - 3.2 防重复提交
- 三、系统监控与告警
- - 1. 分布式链路追踪
  - 2. 监控指标收集
- 四、最佳实践建议
- 五、注意事项

分布式并发场景的核心问题与解决方案

一、核心问题分析

1. 分布式事务问题

在分布式环境下，一个业务操作可能横跨多个服务，比如创建订单时涉及：

订单服务：创建订单
库存服务：扣减库存
支付服务：冻结余额
积分服务：赠送积分

可能出现的问题：

部分服务成功，部分服务失败
网络超时导致事务状态不确定
服务宕机导致事务中断

2. 数据一致性问题

在分布式系统中，由于CAP理论的限制，我们通常需要在一致性和可用性之间做出选择。

典型场景：

主从数据库的数据同步延迟
分布式缓存的数据一致性
跨服务的数据依赖

3. 并发控制问题

多个节点同时处理请求时的并发控制：

超卖问题
重复下单
数据竞争

4. 分布式锁失效问题

Redis主从切换导致锁失效
时钟不同步导致的锁判断错误
网络分区导致的锁状态不一致

二、解决方案

1. 分布式事务解决方案

1.1 可靠消息最终一致性方案

@Service
@Slf4j
public class OrderServiceImpl implements OrderService {@Autowiredprivate RocketMQTemplate rocketMQTemplate;@Autowiredprivate OrderMapper orderMapper;@Autowiredprivate MessageMapper messageMapper;@Transactional(rollbackFor = Exception.class)public void createOrder(OrderDTO orderDTO) {// 1. 事务消息表记录TransactionMessage message = new TransactionMessage();message.setMessageId(UUID.randomUUID().toString());message.setMessage(JSON.toJSONString(orderDTO));message.setStatus(MessageStatus.PREPARING);messageMapper.insert(message);// 2. 创建订单Order order = convertToOrder(orderDTO);orderMapper.insert(order);// 3. 发送事务消息sendTransactionMessage(message);}private void sendTransactionMessage(TransactionMessage message) {Message msg = MessageBuilder.withPayload(message).setHeader("messageId", message.getMessageId()).build();TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction("ORDER_TOPIC",msg,null);if (result.getLocalTransactionState() != LocalTransactionState.COMMIT_MESSAGE) {throw new BusinessException("发送事务消息失败");}}
}

消息消费者实现：

@Component
@RocketMQMessageListener(topic = "ORDER_TOPIC",consumerGroup = "order-consumer-group"
)
public class OrderMessageListener implements RocketMQListener<Message> {@Autowiredprivate StockService stockService;@Autowiredprivate MessageMapper messageMapper;@Overridepublic void onMessage(Message message) {String messageId = message.getHeaders().get("messageId", String.class);// 1. 检查消息是否已处理if (messageMapper.checkProcessed(messageId)) {return;}try {// 2. 处理业务逻辑TransactionMessage txMessage = JSON.parseObject(new String((byte[]) message.getPayload()),TransactionMessage.class);OrderDTO orderDTO = JSON.parseObject(txMessage.getMessage(),OrderDTO.class);// 3. 扣减库存stockService.decreaseStock(orderDTO.getProductId(), orderDTO.getQuantity());// 4. 更新消息状态messageMapper.markAsProcessed(messageId);} catch (Exception e) {// 5. 失败处理messageMapper.markAsFailed(messageId, e.getMessage());// 根据业务需求决定是否抛出异常重试throw e;}}
}

1.2 TCC方案实现

@Service
public class OrderTccServiceImpl implements OrderTccService {@Autowiredprivate OrderMapper orderMapper;@Autowiredprivate StockTccService stockTccService;@Autowiredprivate PaymentTccService paymentTccService;@GlobalTransactionalpublic void createOrder(OrderDTO orderDTO) {// 1. Try阶段// 1.1 订单服务TryOrder order = prepareTryOrder(orderDTO);// 1.2 库存服务TrystockTccService.tryDecrease(orderDTO.getProductId(),orderDTO.getQuantity());// 1.3 支付服务TrypaymentTccService.tryFreeze(orderDTO.getUserId(),orderDTO.getAmount());}// Try阶段的订单处理private Order prepareTryOrder(OrderDTO orderDTO) {Order order = convertToOrder(orderDTO);order.setStatus(OrderStatus.TRY);orderMapper.insert(order);return order;}// Confirm阶段的订单处理public void confirmOrder(BusinessActionContext context) {String orderId = context.getActionContext("orderId").toString();Order order = orderMapper.selectById(orderId);order.setStatus(OrderStatus.CONFIRMED);orderMapper.updateById(order);}// Cancel阶段的订单处理public void cancelOrder(BusinessActionContext context) {String orderId = context.getActionContext("orderId").toString();Order order = orderMapper.selectById(orderId);order.setStatus(OrderStatus.CANCELED);orderMapper.updateById(order);}
}

2. 缓存一致性解决方案

2.1 延迟双删策略

@Service
public class ProductServiceImpl implements ProductService {@Autowiredprivate ProductMapper productMapper;@Autowiredprivate RedisTemplate<String, Product> redisTemplate;@Autowiredprivate ThreadPoolExecutor threadPoolExecutor;private static final String PRODUCT_CACHE_KEY = "product:";private static final long DELAY_DELETE_TIME = 1000; // 1秒@Transactional(rollbackFor = Exception.class)public void updateProduct(Product product) {// 1. 删除缓存String cacheKey = PRODUCT_CACHE_KEY + product.getId();redisTemplate.delete(cacheKey);// 2. 更新数据库productMapper.updateById(product);// 3. 延迟双删threadPoolExecutor.execute(() -> {try {Thread.sleep(DELAY_DELETE_TIME);redisTemplate.delete(cacheKey);} catch (InterruptedException e) {Thread.currentThread().interrupt();log.error("延迟双删失败", e);}});}
}

2.2 Canal方案

@Component
public class ProductCanalClient {@Autowiredprivate RedisTemplate<String, Product> redisTemplate;@Listen(table = "product")public void handleProductChange(CanalEntry.Entry entry) {if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {CanalEntry.RowChange rowChange = entry.getRowChange();for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {if (rowChange.getEventType() == CanalEntry.EventType.UPDATE) {// 处理更新事件handleProductUpdate(rowData);} else if (rowChange.getEventType() == CanalEntry.EventType.DELETE) {// 处理删除事件handleProductDelete(rowData);}}}}private void handleProductUpdate(CanalEntry.RowData rowData) {// 解析变更数据Map<String, String> data = parseRowData(rowData.getAfterColumnsList());String productId = data.get("id");// 更新缓存String cacheKey = "product:" + productId;Product product = convertToProduct(data);redisTemplate.opsForValue().set(cacheKey, product);}
}

3. 并发控制解决方案

3.1 基于Redis的原子操作

@Service
public class StockService {@Autowiredprivate StringRedisTemplate redisTemplate;private static final String STOCK_KEY = "product:stock:";public boolean decreaseStock(Long productId, Integer quantity) {String key = STOCK_KEY + productId;// Lua脚本保证原子性String script = "local stock = redis.call('get', KEYS[1]) " +"if stock and tonumber(stock) >= tonumber(ARGV[1]) then " +"    return redis.call('decrby', KEYS[1], ARGV[1]) " +"end " +"return -1";DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();redisScript.setScriptText(script);redisScript.setResultType(Long.class);Long result = redisTemplate.execute(redisScript,Collections.singletonList(key),quantity.toString());return result != null && result >= 0;}
}

3.2 防重复提交

@Aspect
@Component
public class RepeatSubmitAspect {@Autowiredprivate StringRedisTemplate redisTemplate;@Around("@annotation(repeatSubmit)")public Object around(ProceedingJoinPoint joinPoint, RepeatSubmit repeatSubmit) throws Throwable {HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();String token = request.getHeader("token");String key = getRepeatSubmitKey(joinPoint, token);// 使用Redis的setIfAbsent实现防重boolean isNotRepeat = redisTemplate.opsForValue().setIfAbsent(key,"1",repeatSubmit.interval(),TimeUnit.MILLISECONDS);if (!isNotRepeat) {throw new BusinessException("请勿重复提交");}return joinPoint.proceed();}
}

三、系统监控与告警

1. 分布式链路追踪

@Configuration
public class SleuthConfig {@Beanpublic Sampler defaultSampler() {return Sampler.ALWAYS_SAMPLE;}
}

2. 监控指标收集

@Component
public class DistributedMetrics {@Autowiredprivate MeterRegistry registry;// 记录分布式锁获取情况private Counter lockCounter;private Timer lockTimer;@PostConstructpublic void init() {lockCounter = registry.counter("distributed.lock.acquire");lockTimer = registry.timer("distributed.lock.time");}public void recordLockAcquire(String lockKey, boolean success) {lockCounter.increment();Tags tags = Tags.of("lock_key", lockKey,"success", String.valueOf(success));registry.counter("distributed.lock.acquire", tags).increment();}public void recordLockTime(String lockKey, long timeMillis) {lockTimer.record(timeMillis, TimeUnit.MILLISECONDS);}
}