场景分析

在这里插入图片描述
分布式系统架构中,队列是分布式的,生产端是分布式集群,消费端也是分布式集群.

相当于有多个消费端同时监听队列,同时减库存,写入订单.

面试题:如何处理消息重复消费的问题. 重复消费大部分场景,需要解决的.

引入2个概念来解决: 幂等的业务方法,和消息的分布式锁.

方法的幂等

结论: 一个方法的一次业务逻辑调用和N次调用的结果是一致的,我们称这种方法就是幂等.

一旦重复消费,一定要把消费的业务逻辑方法(orderAdd)设计成幂等的.

幂等的方法
- GET方法: 查询方法,天生幂等.
- DELETE方法: 删除方法,天生幂等.
- PUT方法: 修改,并不是天生幂等,需要设计
  - 减少库存:
    - update stock_tbl set stock=stock-#{stock} where id=#{id}(不是幂等)
    - select * from stock_log where order_id=#{orderId}(查询日志,判断是否已经见过库存了),没有数据 update stock_tbl set stock=stock-#{stock} where id=#{id},insert into stock_log (字段) values (订单id,商品减库存信息) (这样设计就幂等了,依然有问题)
- POST方法: 新增,并不是天生幂等,需要设计
  - 新增订单: insert into order_tbl (order_id,order_item_id,count,user_id) values (各种属性);如果使用唯一属性校验,作用在order_id order_sn(编号).同一张订单,这个字段值是相同(幂等满足,没做幂等不满足)
当前orderAdd方法设计幂等的解决思路(之一)
- 使用订单id或者订单编号,**userId+商品id(**这个只满足当前我们的案例特点,不满足实际场景.)查询订单,如果已经存在了,库存不减少,订单不增了,购物车不用删除了

@Override
public void orderAdd(OrderAddDTO orderAddDTO) {
    //幂等设计思路: 利用userId和commodityCode 查询,如果已经存在了订单,方法直接执行结束
    //如果结果不存在,减库存,生单,删除购物车
    int count=orderMapper.selectExists(orderAddDTO);
    if (count>0){
        log.debug("订单已经新增了");
        return;
    }
    StockReduceCountDTO countDTO=new StockReduceCountDTO();
    countDTO.setCommodityCode(orderAddDTO.getCommodityCode());
    countDTO.setReduceCount(orderAddDTO.getCount());
    // 利用Dubbo调用stock模块减少库存的业务逻辑层方法实现功能
    stockService.reduceCommodityCount(countDTO);
    // 2.从购物车中删除用户选中的商品(调用Cart模块删除购物车中商品的方法)
    // 利用dubbo调用cart模块删除购物车中商品的方法实现功能
    Order order=new Order();
    BeanUtils.copyProperties(orderAddDTO,order);
    // 下面执行新增 假设insert是幂等的.
    orderMapper.insertOrder(order);
    log.info("新增订单信息为:{}",order);
    cartService.cartDelete(orderAddDTO);
}

@Select("select count(id) from " +
"order_tbl where user_id=#{userId} and commodity_code=#{commodityCode}")
int selectExists(OrderAddDTO orderAddDTO);

分布式锁

当前分布式消费架构
在这里插入图片描述
即使,将方法设计成幂等,这个架构中,消息重复消费

,满足线程安全问题的所有因素

并发/多线程
写操作
共享数据

只要解决其中一点,线程安全问题就消失了.

并发多线程–>串行

写操作–> 避免写(不能满足当前案例,必须写)

共享数据–>个体数据(不能满足,重复消费,重复订单是前提)

分布式线程安全问题的解决方案—分布式锁

错误思路: 引入synchronized同步锁,不能解决分布式场景下,多个进程的并发线程安全问题.

概念: 分布式场景下,多进程,多线程并发的抢锁机制. 抢到资源锁,执行业务逻辑,抢不到等待或者放弃执行.能够避免对同一个资源出现并发多线程操作的解决方案.

和synchronized的区别在于 synchronizeds本地锁.管理一个进程中的多线程,分布式锁是管理多个进程中的多线程.

分布式锁当前落地方案: redis setnx命令

Redis实现分布式锁

抢锁的设计思路

目标: 多线程执行业务之前,先判断执行权限,抢锁,抢到锁的才能执行业务,抢不到的不执行.(当前案例中,抢锁,然后执行的业务逻辑是:orderAdd)

抢锁如何执行?: setnx key “”
key值如何设计?: 需要结合业务,设计key值(redis中最主要的功能,都关系到key值的设计),抢锁的逻辑中,满足是业务数据,满足重复消费的重复数据.就可以实现这个key值的设计. 消息Id是重复的.
当前业务流程设计缺陷: 如果有一个消费者抢到锁了,执行了业务方法.执行完成后,没有释放锁的机制.如果引入等待重抢的机制,由于抢到锁的没有释放,会导致死锁.

释放锁的逻辑引入

上述整改的流程中避免了死锁问题,但是存在删除失败导致死锁的问题.

所以,要保证del释放没有成功,在redis也一定不会长期保存.

兜底的解决死锁问题.基本不会出现死锁了.
为了解决误删除的问题,抢锁的时候setnx key value值设计成一个随机数.

随机数两个消费,多个消费者生成相同的可能性极低.

分布式锁案例

@Component
@RocketMQMessageListener(
        topic = "business-order-topic",
        consumerGroup = "${rocketmq.consumer.group}",
        selectorExpression = "orderAdd")
@Slf4j
public class OrderAddConsumerListener implements RocketMQListener<MessageExt> {
    @Autowired
    private IOrderService orderService;
    @Autowired
    private StringRedisTemplate redisTemplate;
    @Override
    public void onMessage(MessageExt msg) {
        //拿到底层消息对象的body
        byte[] body = msg.getBody();
        //尝试先解析成string
        String orderJson=new String(body, StandardCharsets.UTF_8);
        System.out.println(orderJson);
        OrderAddDTO orderAddDTO=
            JSON.toJavaObject(JSON.parseObject(orderJson),OrderAddDTO.class);
        System.out.println(orderAddDTO);
        //1.生成锁的key值,生成当前这把锁的随机数
        //准备锁key
        String msgKeyLock="msg:order:add:"+msg.getMsgId();
        //准备随机数 4 6 8位
        String randCode=new Random().nextInt(9000)+1000+"";
        ValueOperations<String, String> stringOps = redisTemplate.opsForValue();
        try{
            //补充消息消费的抢锁机制
            //2.抢锁 setnx msgKeyLock randCode expire 10s
            Boolean tryLockSuccess = stringOps
                    .setIfAbsent(msgKeyLock, randCode, 10, TimeUnit.SECONDS);
            //3.判断 抢锁成功还是失败
            if(!tryLockSuccess){
                //3.2 失败了 可以等待5秒重新抢锁,也可以直接结束
                //尝试这里使用while编写等待5秒重新抢的逻辑
                log.info("有别人抢锁了,msgKey:{},value:{}",msgKeyLock,randCode);
                return;
            }
            //3.1 成功了 执行orderAdd
            orderService.orderAdd(orderAddDTO);
        }catch (CoolSharkServiceException e){
            //业务异常,说明订单新增业务性失败,比如库存没了
            log.error("库存减少失败,库存触底了:{},异常信息:{}",orderAddDTO,e.getMessage());
        }finally {
            //释放锁 读以下锁的value值,等于当前生成value才释放
            String s = stringOps.get(msgKeyLock);
            if (s!=null && s.equals(randCode)){
                //del msgKeyLock
                redisTemplate.delete(msgKeyLock);
            }
        }
    }
}

直击面试

rocketmq什么时候重复消费

在broker做扩展的时候,消息队列的消息,做扩展的时候,原本存储在原队列的消息,会进行rebalance重平衡.
消费开始阶段
消费者consumer1 所在group1 绑定队列,push消费模式,使得消费者接受到了queue1 queue2的6条消息.消费过程,成功执行,即将返回确认.

总结:
消费者并发消费的逻辑,同一组消费者绑定分布式队列,推送批量的消息
在某个消费者还没有来得及消费,或者没来得及返回确认给rocketmq,队列发生了扩容缩容
rocketmq会对队列中所有的消息做rebalance重平衡(消息重新分配给不同队列),消费者绑定也充平衡
导致已经推送的但是未返回确认的消息,被发送给不同消费者多次.