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

flutter 消息并发时处理,递归查询

收到新消息的时候执行receiveNewConversation方法

可以自己模拟一下两条数据插入,延时执行插入会话的操作
收到一条新的会话消息,先记录会话ID到列表,直到第一条处理完(插入数据库后清理这个会话ID),才处理第二条同会话ID的消息(一直在空递归中)
(递归处理的时候看能一直监听到global的那个conversation列表是否有某个会话的数据,因为在不断重复执行)

核心观念是:1.确保处理多条同会话ID的消息时一定是单线程,2.且插入第一条时要卡住,不要让第二条进来,等第一条插入数据库完成后再执行第二条

注意:如果单纯延时100毫秒再执行第二条同会话ID消息,这种方法依旧是卡不住第二条消息,还是会出现两个会话同时插入数据库,A在12:01执行插入,B在12:02执行插入,A耗时两秒,B耗时一秒,两者都是12:03同时插入消息,导致出现两个会话。

所以还是要加事务锁,第一条执行后锁住,等完成后再执行第二条

  //新会话限制列表,存放会话IDstatic List<String> newConversationList = [];//会话队列static List<WsImMessageQueue> wsImConversationQueue = [];//是否正在处理会话递归static bool wsImConversationQueueState = false;
//不存在会话,则保存会话ID到新会话列表,等插完数据库后再清除新会话列表。
// 第二条会话进来插入到消息列表要等待,等新会话列表没有这个会话ID才把消息插入数据库///收到新会话处理receiveNewConversation(RecvChatMsg recvChatMsg, bool isOffLineMessage) {//会话队列添加数据Global.wsImConversationQueue.add(WsImMessageQueue(recvChatMsg: recvChatMsg, isOffLineMessage: isOffLineMessage));PrintUtil.prints('$TAG 中台IM消息 插入会话消息到wsImConversationQueue: ${utf8.decode(recvChatMsg.body)}   Global.wsImConversationQueue的长度${Global.wsImConversationQueue.length}');if (!Global.wsImConversationQueueState) {Global.wsImConversationQueueState = true;//开始递归startProcessingConversationQueue(true);PrintUtil.prints('$TAG 中台IM消息 递归完成:${Global.wsImConversationQueueState}');}}startProcessingConversationQueue(bool isExistMessage) {if (isExistMessage) {//不为空的时候执行if (Global.wsImConversationQueue.isNotEmpty) {PrintUtil.prints('$TAG 中台IM消息 === > 开始执行递归方法体');//取第一个数据if (Global.wsImConversationQueue[0].recvChatMsg != null) {WsImMessage wsImMessage =addWsImMessage(Global.wsImConversationQueue[0].recvChatMsg);PrintUtil.prints('$TAG 中台IM消息 === > 会话列表第一个会话ID${wsImMessage.conversationId}');//第一条新消息不会被拦截,第二条同会话的消息会被拦截,查询该会话是否在新会话列表里面,如果为true,不给执行,一直执行递归,直到第一条会话插入数据库成功再执行这个判断里面if (checkConversationId(wsImMessage.conversationId ?? '') == false) {PrintUtil.prints('$TAG 中台IM消息 === > 第一个数据或者会话列表第一个会话ID已从新会话限制列表里面移除');WsImDBUtil().insertMessage(Global.wsImConversationQueue[0].recvChatMsg!,Global.wsImConversationQueue[0].isOffLineMessage);//如果下一条进来后还是没有查出来赋值,那还是会拦截不到,一般在一秒内完成,应该影响不大WsImConversationDb.queryId(wsImMessage.conversationId).then((value) {if (value.isEmpty) {if (wsImMessage.conversationId != null) {//插入到新会话限制列表里面Global.newConversationList.add(wsImMessage.conversationId!);PrintUtil.prints('$TAG 中台IM消息 === > 没有该会话就添加会话ID到会话队列:${Global.newConversationList}');}} else {}});PrintUtil.prints('$TAG 中台IM消息 === > 插入消息,新会话限制列表:${Global.newConversationList}');Global.wsImConversationQueue.remove(Global.wsImConversationQueue[0]);PrintUtil.prints('$TAG 中台IM消息 === > 移除消息,消息队列还剩:${Global.wsImConversationQueue.length}');}Future.delayed(const Duration(milliseconds: 1000), () {//1秒后进行递归startProcessingConversationQueue(Global.wsImConversationQueue.isNotEmpty);PrintUtil.prints('$TAG 中台IM消息 === > 递归完成或者下一个递归');});}}} else {PrintUtil.prints('$TAG 中台IM消息 === > 消息处理完成');Global.wsImConversationQueueState = false;}}//判断会话ID是否在新会话列表里面,存在说明要等待这会话ID插入数据完成后才进行其他新会话的插入bool checkConversationId(String id) {bool check = false;for (int i = 0; i < Global.newConversationList.length; i++) {if (id == Global.newConversationList[i]) {check = true;}}PrintUtil.prints('$TAG 中台IM消息 === > 会话列表第一个会话ID在不在新会话限制列表里面:${check}   新会话限制列表:${Global.newConversationList}');return check;}///添加消息内容WsImMessage addWsImMessage(RecvChatMsg? recvChatMsg) {//默认会话ID是:收到谁消息就是谁的会话IDString conId = 'c2c_${recvChatMsg?.fromUserId.toInt()}';//如果发送者是自己,且接收者不是自己,则是服务器下发的消息,会话ID是对方if ('${recvChatMsg?.fromUserId.toInt()}' == Global.userId) {if ('${recvChatMsg?.toUserId.toInt()}' != Global.userId) {conId = 'c2c_${recvChatMsg?.toUserId.toInt()}';}}WsImMessage wsImMessage = WsImMessage(userId: int.parse(Global.userId),conversationId: conId,serverMsgId: recvChatMsg?.serverMsgId.toInt(),fromUserId: recvChatMsg?.fromUserId.toInt(),toUserId: recvChatMsg?.toUserId.toInt(),sendTime: recvChatMsg?.sendTime.toInt(),messageBody: WsImMessageBody.fromJson(utf8.decode(recvChatMsg!.body)),messageState: WsImMessageState.wsImMessageSendSuccess,isSelf: 0,isRead: 0);return wsImMessage;}
http://www.lryc.cn/news/195635.html

相关文章:

  • 第五十八章 学习常用技能 - 查看查询缓存
  • AI 辅助学 Java | 专栏 1 帮你学 Java
  • 2023_Spark_实验十六:编写LoggerLevel方法及getLocalSparkSession方法
  • 彻底搞懂:防止表单重复提交,前端限制还是后端限制?
  • OCPP1.6协议
  • 【数据存储:小端模式和大端模式】
  • 【git】gitlab安装、备份
  • C51--基本认知
  • centos7 安装 mysql 8.0
  • Vue15 计算属性VS监视属性(侦听属性)
  • 快速全面掌握数据库系统核心知识点
  • 学习笔记 | 音视频 | 推流项目框架及细节
  • 拓扑几何学
  • 1.12.C++项目:仿muduo库实现并发服务器之LoopThreadPool模块的设计
  • SpringBoot介绍
  • 2022最新版-李宏毅机器学习深度学习课程-P17 卷积神经网络CNN
  • 微博清理僵尸粉
  • 创建React Native的第一个hello world工程
  • 基础课3——自然语言处理的应用
  • 理解 Git 的三个工作区:工作区、暂存区和版本库
  • web前端基础训练-----创建用户反馈表单
  • Scrum 敏捷管理流程图及敏捷管理工具
  • Android Handler/Looper视角看UI线程的原理
  • 【网络】网络入门
  • GO-实现简单文本格式 文本字体颜色、大小、突出
  • 铅华洗尽,粉黛不施,人工智能AI基于ProPainter技术去除图片以及视频水印(Python3.10)
  • latex,不带行号的algorithm
  • RocketMQ高性能核心原理与源码架构剖析
  • MATLAB中zp2tf函数用法
  • 解决:uniapp项目中调用小程序的chooseAddress() API失效