当前位置： 首页 > news >正文

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

news 2025/9/15 0:49:27

文章目录

  • 分布式并发场景的核心问题与解决方案
    • 一、核心问题分析
      • 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);}
}

四、最佳实践建议

  1. 业务设计层面：

    • 尽量避免复杂分布式事务
    • 考虑业务可补偿性
    • 合理设计重试机制

  2. 技术选型层面：

    • 优先考虑消息队列解耦
    • 合理使用缓存
    • 选择合适的分布式事务方案

  3. 监控运维层面：

    • 完善的监控系统
    • 合理的告警阈值
    • 灾难恢复预案

  4. 性能优化层面：

    • 合理的数据分片策略
    • 避免长事务
    • 批量处理优化

五、注意事项

  1. 数据库层面：

    • 避免大事务
    • 合理设计索引
    • 注意死锁问题

  2. 缓存层面：

    • 防止缓存雪崩
    • 注意缓存穿透
    • 合理设置过期时间

  3. 消息队列层面：

    • 保证消息可靠性
    • 处理重复消息
    • 注意消息顺序性

  4. 分布式锁层面：

    • 防止锁失效
    • 避免死锁
    • 合理设置超时时间
http://www.lryc.cn/news/471216.html

相关文章：

  • D - Many Segments 2（ABC377)
  • 数组指针和指针数组的区别
  • 【VUE点击父组件按钮，跳转到子组件】
  • Java列表排序：方法与实践
  • 哈希及其封装实现unordermap和set
  • 在 AMD GPU 上构建解码器 Transformer 模型
  • Canvas简历编辑器-选中绘制与拖拽多选交互设计
  • 简单工厂（Simple Factory）
  • ffmpeg拉流分段存储到文件-笔记
  • Java 实习工资大概是多少？——解读影响薪资的因素
  • 【Linux】万字详解：Linux文件系统与软硬链接
  • spacenavd
  • C#WPF的XAML的语法详谈和特性
  • 一篇文章讲透数据结构之二叉搜索树
  • 新手入门c++（8）
  • 新手铲屎官提问，有哪几款噪音低的宠物空气净化器推荐
  • 解决RabbitMQ脑裂问题
  • 经纬恒润AUTOSAR成功适配芯钛科技Alioth TTA8车规级芯片
  • 4、java random随机数、一维数组、二维数组
  • C++ 魔法三钥：解锁高效编程的封装、继承与多态
  • 姿态传感器（学习笔记上）
  • labelimg使用教程
  • 力扣21 ： 合并两个有序链表
  • 【Spring】Spring Boot 配置文件（7）
  • 《向量数据库指南》——解锁Wikipedia文章向量的跨语言搜索秘籍
  • 【力扣 + 牛客 | SQL题 | 每日5题】牛客SQL热题204，201，215
  • 下载数据集用于图像分类并自动分为训练集和测试集方法
  • Python xlrd库介绍
  • Javascript立即执行函数
  • Linux相关概念和易错知识点（17）（文件、文件的系统调用接口、C语言标准流）