BirdTalk IM集群中消息流转策略讨论

2024-06-21 16:20

本文主要是介绍BirdTalk IM集群中消息流转策略讨论,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

BirdTalk IM集群中消息流转策略讨论

目前群聊的存储策略是1写多读方案;每个群组一个队列,按时间顺序排列,不区分用户;

私聊的存储是写扩散的,每个人都有自己的消息队列,按时间顺序 保存所有的消息,不区分会话;

1、单机模式

1.1 私聊消息

登录的算法,要求用户按照ID指定的服务器,而不是按照IP,这是因为ip经常发生变动;用ID可以保证各个会话终端都在一个服务器上;

私聊有3类确认:

1)服务器入库的成功回执;

2)对方给的送达回执;

3)阅读回执;

客户端需要一个定时器,对发送的消息监控,如果30秒没有回复成功,认为发送失败;最多尝试3次;

对于没有送达回执的消息,需要向服务区发送查询请求,(回执可能丢失),查询回执会让服务器对在线用户尝试重发;只要客户端实现的正确的,就肯定不会丢消息并应答送达回执;阅读回执可有可无;

1.2 群聊消息

对于单机的群聊,对于内存中没有的用户,就是不在线的,不转发数据;

当群聊用户某个时间点登录,那么先做准备工作,标记自己状态,然后开始同步历史数据,这样能保证数据不丢失。

2、集群模式

不同于TINODE集群直接使用RPC连接,而是使用消息队列同步消息,当一条消息需要从一台服务器路由到另一一台服务器时,可以使用kafka的一个主题来实现,每个服务器一个单独的主题用于接收;这比每对服务器都要建立一个连接好多了,如果集群有1000台服务器,那么就需要1000个TCP连接。

路由可以使用redis辅助实现,为了及时通知对方发现路由变更,也需要使用一个公共的主题来广播消息。

2.1 私聊消息

假如有服务器A和服务器B,小明在服务器A,小美在服务器B;

开始时候小美不在线,消息都写入了小美的消息队列;当某一时刻,小美在服务器B登录,那么:

1)先告知集群当前此用户的此会话在服务器B;

2)执行一次同步历史数据之后,准备接收来自其他服务器用户的数据,当然也包括服务器A;

这里存在一个问题:

服务器A上的小明发现小美上线前,将数据写到了小美的消息队列,但是其实此刻,小美已经上线并且同步完数据了;那么这条消息就会被服务器遗漏,小美客户端也无从得知该条消息;

这个问题的产生主要是路由状态的扩散需要时间,异步的同步历史数据会造成潜在的数据丢失。

但是,与群聊不同,私聊是有回执的;当小明发现小美一直都没有回复接收回执,则会一直查询回执,服务器A会重发消息,此时如果发现了消息路由在服务器B,可以保证消息重发而不丢失;

但是有个缺点也很明显:有可能之前丢失的消息会在客户端出现乱序!!!

2.2 群聊消息

假如有服务器A和服务器B,一个群组G100;小明在服务器A,小美在服务器B;

当小美在服务器登录时间不确定的时候如何保证群聊消息不丢失?

可能会丢失的原因在于:

1)分布式环境下,消息的编号不是连续的流水号,无法通过编号来判断是否发生了丢失;

2)登录后如果只与服务器B同步一次数据,同步数据与其他服务器发现路由有时间差,时间差内的消息会丢失;

那么处理方式也很明显:

**1)传统轮询模式:**每次查询都告知服务器最后一条的ID,然后从数据库表中查询后续的消息,对于scylladb来说肯定是比传统的数据库强,但是效率仍然很低;

**2)暴力的集群同步:**不管服务器B是否有G100群组的用户在线,一直连续不断的转发消息,那么可以服务器B上存在的消息在时间上是连续的,那么登录后,同步一次数据就可以保证与后续转发的数据无缝衔接;

2.3 优化解决方法

那么针对2种方案时候有优化的方案呢?矫正后的算法对私聊和群聊都有效,原理是执行有限次同步以便在时间上覆盖所有范围:

3) 时间戳矫正法:

小美在服务器B登录,就绪后将状态广播到集群,之后开始同步群G100的历史数据,记录下同步到的最后一条消息的ID;假设同步的最新一条数据编号LAST_ID=1000200;(真实环境使用雪花算法)

这个LAST_ID本质上记录的是同步点;同步点仅仅随着同步查询动作而更新;

可能一:服务器A早就发现服务器B有群组G100其他用户,一直在转发数据,那么小美不会丢失数据,因为是流程是先设置状态后同步数据,服务器B可以保证消息没有丢失;对于同步前收到的数据,客户端先写到本地库,同步数据完成后再显示就不会乱序;

可能二:

小美开始是离线状态,等到小美上线服务器B,将状态发布到集群;服务器A收到群组G100到服务器B的路由;此时服务器A针对此路由,应当记录一个发现路由的时间戳(START_TM),并在在此后所有针对此路由的转发的消息都需要带上这个时间戳START_TM,含义是从此时刻起开始转发消息;(加入针对这个路由发送的第一个消息ID为 1000500,这个时间戳可以设置为发现路由后转发的第一条消息ID);

服务器B收到消息后,比对时间戳,如果 LAST_ID >= START_TM 则说明同步的的时间范围与开始转发的时间没有时间差;(在可能一中,也是满足这个不等式)。

如果 LAST_ID < START_TM 则说明存在时间差,有丢失消息风险,应该按照范围再1次执行同步数据,这样就能保证消息不丢失。同步消息后更新LAST_ID,含义是同步点可以标记到当前位置。

后续,假如有服务器C转发过来数据,也是同样大道理: 因为消息ID虽然不连续,但是雪花算法单调递增,而且里面包含了时间戳;

只要同步点晚于路由发现时间,就可以保证消息没有遗漏。

在这里插入图片描述

当服务器B上所有的关于群组G100的用户都离线后,则不需要服务器A转发数据了,则删除该路由。

相关代码:

// 发送方
type RouteData struct{MemCount int64  // 当前服务器上群组活动人员数量StartTm  int64  // 发现路由的时间戳,
}// 在群组中添加2个映射记录数据
type Group struct{GroupId  int64...RouteMap     map[string]*RouteData  // 发方:发现到某路由时候设置时间戳}
// 收方:
// 由于用户支持多终端同时在线,所以路由其实记录的用户的会话,而不是用户本身,有可能出现同一个用户登录到不同的主机上;
type Session struct{Sid  int64...GroupSynId map[int64]int64        // 收方:每个组同步操作得到的位置:groupId ->  LAST_ID
}

当服务器B有群组G100的用户时,会记录如下

//lock
group.RouteMap["serverB"] = &RouteData{MemCount: 1,StartTm:  1000500}

当路由消失后,则删除此路由;

针对私聊采用类似的策略:

其中,有可能出现同一个用户登录到不同的主机上;

type SessionOnserverData struct{    // 每个会话在哪个服务器上,记录首次发现路由的时间SessionList  []int64StartTm int64
}// 有可能该好友的会话分布在多种终端,但是按照算法大多在同一个服务器
type UserRouteData struct{SessionOnServerMap  map[string]*SessionOnserverData   // 服务器——>会话列表}type User struct{UserId int64...FriendRoute map[int64]*UserRouteData  // 针对好友的路由,UID-> 路由信息
}
// 接收方,在会话中记录好友发来的最后的消息的LAST_ID,
type Session struct{Sid  int64..LastSynPoint int64  //会话已经完成同步最新的ID,因为所有消息都存在在一起,不用区分好友
}

从上面可以总结:

收方记录的同步点是基于队列的;

群聊的路由发现是以服务器为单位执行转发;(减少处理的复杂度)

私聊的路由发现是基于服务器为单位执行转发;

2.4 收方消息处理

收方如果从消息队列拿到数据后,直接在线程池中处理,会造成用户数据的并发冲突,我们这里针对每个用户需要线性处理;

合适的方式是将消息分发到各个会话的消息缓存队列中,由会话的读协程来处理这个内容;

完。

这篇关于BirdTalk IM集群中消息流转策略讨论的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1081720

相关文章

SpringKafka消息发布之KafkaTemplate与事务支持功能

《SpringKafka消息发布之KafkaTemplate与事务支持功能》通过本文介绍的基本用法、序列化选项、事务支持、错误处理和性能优化技术,开发者可以构建高效可靠的Kafka消息发布系统,事务支... 目录引言一、KafkaTemplate基础二、消息序列化三、事务支持机制四、错误处理与重试五、性能优

SpringIntegration消息路由之Router的条件路由与过滤功能

《SpringIntegration消息路由之Router的条件路由与过滤功能》本文详细介绍了Router的基础概念、条件路由实现、基于消息头的路由、动态路由与路由表、消息过滤与选择性路由以及错误处理... 目录引言一、Router基础概念二、条件路由实现三、基于消息头的路由四、动态路由与路由表五、消息过滤

SpringBoot如何通过Map实现策略模式

《SpringBoot如何通过Map实现策略模式》策略模式是一种行为设计模式,它允许在运行时选择算法的行为,在Spring框架中,我们可以利用@Resource注解和Map集合来优雅地实现策略模式,这... 目录前言底层机制解析Spring的集合类型自动装配@Resource注解的行为实现原理使用直接使用M

MySQL中闪回功能的方案讨论及实现

《MySQL中闪回功能的方案讨论及实现》Oracle有一个闪回(flashback)功能,能够用户恢复误操作的数据,这篇文章主要来和大家讨论一下MySQL中支持闪回功能的方案,有需要的可以了解下... 目录1、 闪回的目标2、 无米无炊一3、 无米无炊二4、 演示5、小结oracle有一个闪回(flashb

Redis 内存淘汰策略深度解析(最新推荐)

《Redis内存淘汰策略深度解析(最新推荐)》本文详细探讨了Redis的内存淘汰策略、实现原理、适用场景及最佳实践,介绍了八种内存淘汰策略,包括noeviction、LRU、LFU、TTL、Rand... 目录一、 内存淘汰策略概述二、内存淘汰策略详解2.1 ​noeviction(不淘汰)​2.2 ​LR

Spring Boot整合消息队列RabbitMQ的实现示例

《SpringBoot整合消息队列RabbitMQ的实现示例》本文主要介绍了SpringBoot整合消息队列RabbitMQ的实现示例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的... 目录RabbitMQ 简介与安装1. RabbitMQ 简介2. RabbitMQ 安装Spring

springboot rocketmq配置生产者和消息者的步骤

《springbootrocketmq配置生产者和消息者的步骤》本文介绍了如何在SpringBoot中集成RocketMQ,包括添加依赖、配置application.yml、创建生产者和消费者,并展... 目录1. 添加依赖2. 配置application.yml3. 创建生产者4. 创建消费者5. 使用在

如何通过Python实现一个消息队列

《如何通过Python实现一个消息队列》这篇文章主要为大家详细介绍了如何通过Python实现一个简单的消息队列,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录如何通过 python 实现消息队列如何把 http 请求放在队列中执行1. 使用 queue.Queue 和 reque

Deepseek使用指南与提问优化策略方式

《Deepseek使用指南与提问优化策略方式》本文介绍了DeepSeek语义搜索引擎的核心功能、集成方法及优化提问策略,通过自然语言处理和机器学习提供精准搜索结果,适用于智能客服、知识库检索等领域... 目录序言1. DeepSeek 概述2. DeepSeek 的集成与使用2.1 DeepSeek API

Redis的数据过期策略和数据淘汰策略

《Redis的数据过期策略和数据淘汰策略》本文主要介绍了Redis的数据过期策略和数据淘汰策略,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一... 目录一、数据过期策略1、惰性删除2、定期删除二、数据淘汰策略1、数据淘汰策略概念2、8种数据淘汰策略