为自己搭建一个分布式 IM(即时通讯) 系统

Posted 2023-04-25

参考技术A

CIM(CROSS-IM) 一款面向开发者的 IM(即时通讯)系统；同时提供了一些组件帮助开发者构建一款属于自己可水平扩展的 IM 。

借助 CIM 你可以实现以下需求：

下面来看看具体的架构设计。

整体主要由以下模块组成：

cim-server

IM 服务端；用于接收 client 连接、消息透传、消息推送等功能。

支持集群部署。

cim-forward-route

消息路由服务器；用于处理消息路由、消息转发、用户登录、用户下线以及一些运营工具（获取在线用户数等）。

cim-client

IM 客户端；给用户使用的消息终端，一个命令即可启动并向其他人发起通讯（群聊、私聊）；同时内置了一些常用命令方便使用。

整体的流程也比较简单，流程图如下：

所以当我们自己部署时需要以下步骤：

接下来重点看看具体的实现，比如群聊、私聊消息如何流转；IM 服务端负载均衡；服务如何注册发现等等。

IM 服务端

先来看看服务端；主要是实现客户端上下线、消息下发等功能。

首先是服务启动：

由于是在 SpringBoot 中搭建的，所以在应用启动时需要启动 Netty 服务。

从 pipline 中可以看出使用了 Protobuf 的编解码（具体报文在客户端中分析）。

注册发现

需要满足 IM 服务端的水平扩展需求，所以 cim-server 是需要将自身数据发布到注册中心的。

所以在应用启动成功后需要将自身数据注册到 Zookeeper 中。

最主要的目的就是将当前应用的 ip + cim-server-port+ http-port 注册上去。

上图是我在演示环境中注册的两个 cim-server 实例（由于在一台服务器，所以只是端口不同）。

这样在客户端（监听这个 Zookeeper 节点）就能实时的知道目前可用的服务信息。

登录

当客户端请求 cim-forward-route 中的登录接口（详见下文）做完业务验证（就相当于日常登录其他网站一样）之后，客户端会向服务端发起一个长连接，如之前的流程所示：

这时客户端会发送一个特殊报文，表明当前是登录信息。

服务端收到后就需要将该客户端的 userID 和当前 Channel 通道关系保存起来。

同时也缓存了用户的信息，也就是 userID 和 用户名。

离线

当客户端断线后也需要将刚才缓存的信息清除掉。

同时也需要调用 route 接口清除相关信息（具体接口看下文）。

IM 路由

从架构图中可以看出，路由层是非常重要的一环；它提供了一系列的 HTTP 服务承接了客户端和服务端。

目前主要是以下几个接口。

注册接口

由于每一个客户端都是需要登录才能使用的，所以第一步自然是注册。

这里就设计的比较简单，直接利用 Redis 来存储用户信息；用户信息也只有 ID 和 userName 而已。

只是为了方便查询在 Redis 中的 KV 又反过来存储了一份 VK，这样 ID 和 userName 都必须唯一。

登录接口

这里的登录和 cim-server 中的登录不一样，具有业务性质，

为了实现只能一个用户登录，使用了 Redis 中的 set 来保存登录信息；利用 userID 作为 key ，重复的登录就会写入失败。

获取一台可用的路由实例也比较简单：

当然要获取 Zookeeper 中的服务实例前自然是需要监听 cim-server 之前注册上去的那个节点。

具体代码如下：

也是在应用启动之后监听 Zookeeper 中的路由节点，一旦发生变化就会更新内部缓存。

群聊接口

这是一个真正发消息的接口，实现的效果就是其中一个客户端发消息，其余所有客户端都能收到！

流程肯定是客户端发送一条消息到服务端，服务端收到后在上文介绍的 SessionSocketHolder 中遍历所有 Channel（通道）然后下发消息即可。

服务端是单机倒也可以，但现在是集群设计。所以所有的客户端会根据之前的轮询算法分配到不同的 cim-server 实例中。

因此就需要路由层来发挥作用了。

路由接口收到消息后首先遍历出所有的客户端和服务实例的关系。

路由关系在 Redis 中的存放如下：

由于 Redis 单线程的特质，当数据量大时；一旦使用 keys 匹配所有 cim-route:* 数据，会导致 Redis 不能处理其他请求。

所以这里改为使用 scan 命令来遍历所有的 cim-route:*。

接着会挨个调用每个客户端所在的服务端的 HTTP 接口用于推送消息。

在 cim-server 中的实现如下：

cim-server 收到消息后会在内部缓存中查询该 userID 的通道，接着只需要发消息即可。

在线用户接口

这是一个辅助接口，可以查询出当前在线用户信息。

实现也很简单，也就是查询之前保存 ”用户登录状态的那个去重 set “即可。

私聊接口

之所以说获取在线用户是一个辅助接口，其实就是用于辅助私聊使用的。

一般我们使用私聊的前提肯定得知道当前哪些用户在线，接着你才会知道你要和谁进行私聊。

类似于这样：

在我们这个场景中，私聊的前提就是需要获得在线用户的 userID。

所以私聊接口在收到消息后需要查询到接收者所在的 cim-server 实例信息，后续的步骤就和群聊一致了。调用接收者所在实例的 HTTP 接口下发信息。

只是群聊是遍历所有的在线用户，私聊只发送一个的区别。

下线接口

一旦客户端下线，我们就需要将之前存放在 Redis 中的一些信息删除掉（路由信息、登录状态）。

IM 客户端

客户端中的一些逻辑其实在上文已经谈到一些了。

登录

第一步也就是登录，需要在启动时调用 route 的登录接口，获得 cim-server 信息再创建连接。

登录过程中 route 接口会判断是否为重复登录，重复登录则会直接退出程序。

接下来是利用 route 接口返回的 cim-server 实例信息（ip+port）创建连接。

最后一步就是发送一个登录标志的信息到服务端，让它保持客户端和 Channel 的关系。

自定义协议

上文提到的一些登录报文、真正的消息报文这些其实都是在我们自定义协议中可以区别出来的。

由于是使用 Google Protocol Buffer 编解码，所以先看看原始格式。

其实这个协议中目前一共就三个字段：

目前主要是三种类型，分别对应不同的业务：

心跳

为了保持客户端和服务端的连接，每隔一段时间没有发送消息都需要自动的发送心跳。

目前的策略是每隔一分钟就是发送一个心跳包到服务端：

这样服务端每隔一分钟没有收到业务消息时就会收到 ping 的心跳包：

内置命令

客户端也内置了一些基本命令来方便使用。

比如输入 :q 就会退出客户端，同时会关闭一些系统资源。

当输入 :olu(onlineUser 的简写)就会去调用 route 的获取所有在线用户接口。

群聊

群聊的使用非常简单，只需要在控制台输入消息回车即可。

这时会去调用 route 的群聊接口。

私聊

私聊也是同理，但前提是需要触发关键字；使用 userId;;消息内容 这样的格式才会给某个用户发送消息，所以一般都需要先使用 :olu 命令获取所以在线用户才方便使用。

消息回调

为了满足一些定制需求，比如消息需要保存之类的。

所以在客户端收到消息之后会回调一个接口，在这个接口中可以自定义实现。

因此先创建了一个 caller 的 bean，这个 bean 中包含了一个 CustomMsgHandleListener 接口，需要自行处理只需要实现此接口即可。

自定义界面

由于我自己不怎么会写界面，但保不准有其他大牛会写。所以客户端中的群聊、私聊、获取在线用户、消息回调等业务(以及之后的业务)都是以接口形式提供。

也方便后面做页面集成，只需要调这些接口就行了；具体实现不用怎么关心。

cim 目前只是第一版，BUG 多，功能少（只拉了几个群友做了测试）；不过后续还会接着完善，至少这一版会给那些没有相关经验的朋友带来一些思路。

欢迎工作一到五年的Java工程师朋友们加入Java程序员开发： 721575865

群内提供免费的Java架构学习资料（里面有高可用、高并发、高性能及分布式、Jvm性能调优、Spring源码，MyBatis，Netty,Redis,Kafka,Mysql,Zookeeper,Tomcat,Docker,Dubbo,Nginx等多个知识点的架构资料）合理利用自己每一分每一秒的时间来学习提升自己，不要再用"没有时间“来掩饰自己思想上的懒惰！趁年轻，使劲拼，给未来的自己一个交代！

IM即时通讯开发可靠的亿级IM消息投递机制

即时通讯（IM）系统最基础、最重要的是消息的及时性与准确性，及时体现在延迟，准确则具体表现为不丢、不重、不乱序。

综合考虑业务场景、系统复杂度、网络流量、终端能耗等，我们的亿级分布式IM消息系统精心设计了消息收发机制，并不断打磨优化，形成了现在的消息可靠投递机制。

整体思路就是：

    1）客户端、服务端共同配合，互相补充；

    2）采用多重机制，从不同层面保障；

    3）拆分上下行，分别处理。

一个完整的IM消息交互逻辑，通常会为两段：

    1）消息上行段：即由消息发送者通过IM实时通道发送给服务端；

    2）消息下行段：由服务端按照一定的策略送达给最终的消息接收人。

消息上行段主要就是依赖IM的实时通道将消息传递给服务端。

这个阶段的消息可靠投递，需要从协议层进行保证，协议层需要提供可靠、有序的双向字节流传输，我们是通过自研的通信协议 RMTP（即 RongCloud Message Transfer Protocol）实现的。

客户端与服务端之间使用长连接，基于 RMTP 协议传输数据。

经过总结，消息下行段主要有三种行为。

1）客户端主动拉取消息，主动拉取有两个触发方式：

    ① 拉取离线消息：与 IM 服务新建立连接成功，用于获取不在线的这段时间未收到的消息；

    ② 定时拉取消息：在客户端最后收到消息后启动定时器，比如 3-5 分钟执行一次。主要有两个目的，一个是用于防止因网络、中间设备等不确定因素引起的通知送达失败，服务端客户端状态不一致，一个是可通过本次请求，对业务层做状态机保活。

2）服务端主动-发送消息（直发消息）：

这是在线消息发送机制之一，简单理解为服务端将消息内容直接发送给客户端，适用于消息频率较低，并且持续交互，比如二人或者群内的正常交流讨论。

3）服务端主动-发送通知（通知拉取）：

这是在线消息发送机制之一，简单理解为服务端给客户端发送一个通知，通知包含时间戳等可作为排序索引的内容，客户端收到通知后，依据自身数据，对比通知内时间戳，发起拉取消息的流程。

这种场景适用于较多消息传递：比如某人有很多大规模的群，每个群内都有很多成员正在激烈讨论。通过通知拉取机制，可以有效的减少客户端服务端网络交互次数，并且对多条消息进行打包，提升有效数据载荷。既能保证时效，又能保证性能。

在上行过程保证发送消息顺序，为了保证消息有序， 最好的方式是按照 userId 区分，然后使用时间戳排序。那么分布式部署情况下，将用户归属到固定的业务服务器上（PS：指的是同一账号的不同端固定连接到相同的业务服务器上），会使得上行排序变得更容易。同时归属到同一个服务器，在多端维护时也更容易。

客户端连接过程：

    1）客户端通过 APP server ，获取到连接使用的 token；

    2）客户端使用 token 通过导航服务，获取具体连接的 IM 接入服务器(CMP)，导航服务通过 userId 计算接入服务器，然后下发，使得某一客户端可以连接在同一台接入服务器(CMP)。

小结一下就是：客户端发出消息后，通过接入服务，按照 userId 投递到指定消息服务器，生成消息 Id， 依据最后一条消息时间，确认更新当前消息的时间戳（如果存在相同时间戳则后延）。然后将时间戳，以及消息 Id，通过 Ack 返回给客户端 ; 然后对上行消息使用 userId + 时间戳进行缓存以及持久化存储，后续业务操作均使用此时间戳。

消息节点在处理完上行流程后，消息按照目标用户投递到所在消息节点，进入下行流程。

下行过程，按照目标 userId 以及本消息在上行过程中生成的时间戳，计算是否需要更新时间戳（正向）。

如果需要更新则对时间戳进行加法操作，直到当前用户时间戳不重复。

如此处理后，目标用户的存储以及客户端接收到消息后的排重可以做到一致，并且可以做到同一个会话内的时间戳是有序的。从而保证同一个接收用户的消息不会出现乱序。

至此：我们已经介绍完了消息的下行交互过程，消息下行过程中的具体实现方式并不简单，以下将详细展开。

1）直发消息：

即服务端主动发送（给目标客户端）的消息：

    1）客户端 SDK 依据本地存储的最新消息时间戳判断，用来做排序等逻辑；

    2）对同一个用户直发消息1条，其他转通知。通知拉取时候客户端选择本地最新一条消息时间戳作为开始拉取时间；

    3）在消息发送过程中，如果上一条消息发送流程未结束，下一条消息则不用直发(s_msg)，而是用通知(s_ntf)。

即服务端主动发送通知（给目标客户端）：

    1）服务端在通知体中携带当前消息时间戳。投递给客户端；

    2）客户端收到通知后，比对本地消息时间戳，选择是否发拉取消息信令；

    3）服务端收到拉取消息信令后，以信令携带的时间戳为开始，查询出消息列表(200 条或者 5M)，并给客户端应答；

    4）客户端收到后，给服务端 ack，服务端维护状态；

    5）客户端拉取消息时使用的时间戳，是客户端本地最新一条消息的时间戳。

具体逻辑是：

    1）用户多个终端链接成功后，发送一条消息，这个消息到达 CMP(IM 接入服务) 后，CMP 做基础检查，然后获此用户的其他终端连接；

    2）服务把客户端上行的消息，封装为服务端下行消息，直接投递给用户的其他客户端。这样完成了发送方的多端抄送，然后将这条消息投递到 IM 服务。进入正常发送投递流程。即时通讯聊天软件app开发可以加蔚可云的v：weikeyun24咨询

针对上面的第2）点，发送方的多端同步没有经过 IM Server，这么做的好处是：

    1）比较快速；

    2）经过越少的服务节点，出问题的几率越小。

接收方多端同步

具体逻辑是：

    1）IM 服务收到消息后，先判断接收方的投递范围，这个范围指的是接收方用户的哪些的终端要接收消息；

    2）IM 服务将范围以及当前消息，发送到 CMP，CMP 依据范围，匹配接收方的终端，然后投递消息。

接收方多端消息同步范围的应用场景，一般都是针对所有终端。

但有一些特殊业务：比如我在 A 客户端上，控制另外某个端的状态，可能需要一些命令消息， 这时候需要这个作用范围，针对性的投递消息。