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

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）消费消息

 @Overridepublic 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存储进去，方便后续进行updatechatDto.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);//设置过滤应该有的tokenRoseFeignConfig.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;}