IM聊天系统为什么需要做消息幂等？如何使用Redis以及Lua脚本做消息幂等【第12期】

2024-02-13 13:20:02
开发
35

0前言

消息收发模型
在这里插入图片描述

在这里插入图片描述
第一张图是一个时序图，第二张图是一个标清楚步骤的流程图，更加清晰。消息的插入环节主要在2步。save部分。主要也是对这个部分就行消息幂等的操作。

前情提要：使用Redis发布 token 以及lua脚本来共同完成消息的幂等

目前已经写的文章有。并且有对应视频版本。
git项目地址【IM即时通信系统（企聊聊）】点击可跳转
 sprinboot单体项目升级成springcloud项目【第一期】
前端项目技术选型以及页面展示【第二期】
分布式权限 shiro + jwt + redis【第三期】
给为服务添加运维模块统一管理【第四期】
微服务数据库模块【第五期】
netty与mq在项目中的使用(第六期)】
分布式websocket即时通信(IM)系统构建指南【第七期】
分布式websocket即时通信(IM)系统保证消息可靠性【第八期】
分布式websocket IM聊天系统相关问题问答【第九期】
什么？websocket也有权限！这个应该怎么做？【第十期】
分布式ID是什么，以美团Leaf为例改造融入自己项目【第十一期】

1.我开源项目IM重复的原因

IM系统中有三个常见的指标。消息可靠（不丢消息）。就是消息不能重复（不重复）。保证消息的时序性（不乱序）。这三个指标非常重要。
消息可靠主要通过报文协议等操作来完成。前面视频有一期讲过报文协议。目前主要采取上述方式去保证消息的可靠性。然后再保证消息可靠性的过程中，有一些需要重试的操作。可能会导致数据库多次插入。需要我们来保证一下消息的幂等。通俗的讲就是保证消息的不重复。

1.客户端会重复的发送消息

客户端是一个timer的机制。客户端a发送给b消息的时候，在0.5秒没有收到b的ack的时候会重发消息，重发三次还没有收到ack视为重发失败
//使用timer机制检测队列里面是否存在ack，如果存在，则超时重发以及限制次数

伪代码如下。用户在线并且不是重试消息的时候，添加到队列里面。

 if (res.params.online == true && res.params.isretry == "false") {
   
            state.queue.offer(state.tempSendMsg);
            //
            //使用timer机制 检测队列里面是否存在ack，如果存在，则超时重发以及限制次数
            const result = await retry(fetchDataFn, 3, 1000, res.params.msgid);
            //三次之后消息还没有发送成功 提示消息发送失败
            if (result == false) {
   
              Toast("消息发送失败，请重新发送");
            }
          } else {
   
            console.log("【IM日志】 接受消息者没有登录或者是重试消息 ");
          }

进行重试的js代码

//重试的一个方法
export function retry(fn, maxRetry, timeout,msg) {
   
  return new Promise(async (resolve, reject) => {
   
    let retryCount = 0;
    let timer;
 
    const run = async () => {
   
      try {
   
        const result = await fn(msg);
        resolve(result);
      } catch (err) {
   
        if (retryCount < maxRetry) {
   
          retryCount++;
          clearTimeout(timer);
          timer = setTimeout(run, timeout);
        } else {
   
          reject(err);
        }
      }
    };
 
    timer = setTimeout(run, timeout);
  });
}

2.mq出现超时等的重试机制

参考上述逻辑图，消息落库的时候异步分发到了mq上面。rocketmq有超时重试机制，会自动重试。导致消息被多次消费。（明天补充个图片例子）

2.如何解决的幂等

为什么要解决幂等，什么情况下出现幂等（明天写）；

使用redis做的幂等。redis做幂等其实有两种思路。

一种思路是我目前正在使用的防重 Token 令牌思路。另一种是下游传递唯一请求编号。主要说明防重token令牌的思路。其实差别就是一个redis里面的键被删除了。另一个没有删除。
防重token令牌
在这里插入图片描述
下游传递唯一请求编号如下

当客户端请求分布式id的时候将其存入redis。也就是获取一个唯一id。当进行消费消息的时候。先判断唯一id在不在。在的话删除redis中的唯一id并且进行业务操作。不再的话就不能进行业务操作来实现的幂等。
流程代码如下所示：

1.获取token以及存储token到redis中；
在loginUser 用户中心服务中

    @RequestMapping(value = "/api/segment/get/{key}")
    public GenericResponse getSegmentId(@PathVariable("key") String key) {
   
        String leafno = get(key, segmentService.getId(key));
        SetOperations<String, String> opsForSet = stringRedisTemplate.opsForSet();
        Long add = opsForSet.add(RedisPrefix.LEAF_PERFIX, leafno);//往集合添加元素
        /**
         * 设置一个10分钟的有效期
         */
//        stringRedisTemplate.expire(RedisPrefix.LEAF_PERFIX,600, TimeUnit.SECONDS);
        return GenericResponse.response(ServiceError.NORMAL,leafno );
    }

我们使用了美团的分布式id来生成分布式id。
2.前台发送消息的时候携带上唯一id

const sendMsg2 = async () => {
   
      const {
    content, toUser } = state;
      const no = await getLeaf();
      let data = {
   
        // 1代表着私聊的意思
        type: 1,
        params: {
   
          msgid: no.content,
          toMessageId: toUser.openid,
          message: content,
          fileType: 0,
          isretry: false,
        },
      };
      if (state.current == 2) {
   
        data = {
   
          type: 9,
          params: {
   
            toMessageId: state.groupId,
            message: content,
            fileType: 0,
          },
        };
      }
      console.log(data);
      state.tempSendMsg = data;
      state.socketServe.send(data);
      state.recesiveAllMsg.push({
   
        type: "self",
        content: content,
      });
      state.content = "";
    };

这个是发送消息的操作
const no = await getLeaf();这行代码请求后端接口。然后构造消息体。
3.聊天服务(Netty)收到前台消息后 mq异步发送消息

 public void sendMessage(String topic ,ChannelHandlerContext ctx, String message, String toUser, String state, Boolean type, String msgid,String token) {
   
        MqMessage messageMQ = new MqMessage();
        messageMQ.setFromId(SessionUtils.getUser(ctx.channel()).getOpenid());
        messageMQ.setToId(toUser);
        messageMQ.setType(state);
        messageMQ.setInfoContent(message);
        messageMQ.setTime(new DateTime().toString());
        messageMQ.setState(type);
        messageMQ.setMsgid(msgid);
        messageMQ.setToken(token);
        messageDispatchService.sendForSave(topic,messageMQ);
    }

发送给保存的主题
4.业务模块（frist）消费消息

 @Override
    public void onMessage(String o) {
   
        String mqmsg =o;
        log.info("RocketMqConsumerService=====消费消息:"+mqmsg);
        //消息内容

        MqMessage message1 = JSON.parseObject(mqmsg, MqMessage.class);
        try {
   
            ChatDto chatDto = new ChatDto();
            chatDto.setContent(message1.getInfoContent());
            chatDto.setToOpenid(message1.getToId());
            chatDto.setGroup(message1.getState());
            //将msgid存储进去，方便后续进行update
            chatDto.setMsgId(message1.getMsgid());
            SetOperations<String, String> opsForSet = stringRedisTemplate.opsForSet();
//            Boolean member = opsForSet.isMember(RedisPrefix.LEAF_PERFIX, message1.getMsgid());
            if( executeOperation(message1.getMsgid())){
   
//                Long remove = opsForSet.remove(RedisPrefix.LEAF_PERFIX, message1.getMsgid());//删除元
                if (message1.getState() !=null){
   
                    if(message1.getType().equals("onLine")){
   
                        /**
                         * 用户在线需要去推送一下
                         */
                        yanUserChatService.saveChat(message1.getFromId(),chatDto,1);
                        SendRequest send = buildSendRequest(message1);
                        //设置过滤应该有的token
                        RoseFeignConfig.token.set(message1.getToken());
                        nettyMqFeign.send(send);
                    }else {
   
                        /**
                         * 离线消息直接落库就链路就结束了
                         */
                        yanUserChatService.saveChat(message1.getFromId(),chatDto,0);
                    }
                }
            }
        }catch (Exception e){
   
            //失败的话需要把redis的这个消息还回去.
            SetOperations<String, String> opsForSet = stringRedisTemplate.opsForSet();
            Long add = opsForSet.add(RedisPrefix.LEAF_PERFIX,  message1.getMsgid());//往集合添加元素
            log.error("consumeMsg 消费mq消息失败.",e);
            // 处理失败，抛出异常，消息会根据重试策略稍后重新消费
            throw new RuntimeException("处理消息时发生错误，消息将被重新消费。");
        }

    }

lua表达式
目前使用redis的类型是set，键是yan_leaf

    /**
     * 幂等的方法,判断list存不存在。存在的话直接删除，下次进来就不存在了。
     * @param token
     * @return
     */
    public boolean executeOperation(String token) {
   
        // Lua脚本
        String script = "if redis.call('sismember', KEYS[1], ARGV[1]) == 1 then return redis.call('srem', KEYS[1], ARGV[1]) else return 0 end";
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>(script, Long.class);

        // 执行Lua脚本
        Long result = stringRedisTemplate.execute(redisScript, Collections.singletonList(RedisPrefix.LEAF_PERFIX), token);

        // 根据Lua脚本执行结果判断操作是否执行
        return result != null && result > 0;
    }

通过这个lua防止并发请求进来导致幂等失败

原文地址:https://blog.csdn.net/qq_21561833/article/details/136104296 本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：https://www.suanlizi.com/kf/1757273400595845120.html 如若内容造成侵权/违法违规/事实不符，请联系《酸梨子》网邮箱：1419361763@qq.com进行投诉反馈，一经查实，立即删除！

阅读全部