我们开始讲goim的长连接网关
这个项目其实是
学Go有一段时间之后觉得应该搞一个什么项目来练练手
当时其实对im就比较好奇
尤其像QQ微信这种他们怎么去实现的
所以当时搜了很多像QQ 微信 Facebook的message
这些服务的他的一些实践和资料
然后基于此
首先考虑就是海量连接的一个怎么去管理
怎么去做消息存储
所以就有了这个项目的一个早期的一个初步想法
然后就开始做开源了
当时构思整个项目的时候
完全是自己假想的一种场景
如果有像QQ微信这样大的一个用户群怎么去设计
所以我们先看一下简单看一下
TCP长连接编程的一些概览
这个系统是怎么设计的
首先的话我们肯定是有一个client 有个客户端
因为可能当时想法是说要做实时通讯
可能使用长连接会比较简单一点
我当时就是想基于websocket
或者是TCP去进行一个长连接
连到我们的comet这个服务
它是一个长连接层
主要是支持websocket和TCP
然后它监听外网的TCP或者是website端口
然后它也支持通过类似像设备id
或者说叫用户信息进行绑定
一个连接通道
我在内部抽象为叫channel
就是客户端和服务端建立连接之后
我希望他们之间的一个会话或者说
很多指令操作像一个通道一样
所以我当时是进行一个channel的抽象
然后comet还支持房间
因为我们有一个场景就是像直播的聊天室这种场景
所以它有两个容器
一个就是channel
channel其实是可以认为是一个设备到我们的comet服务的一个通道
第二个叫room房间
就是说一个人他可以在可以在不同的房间里面
这是第一个模块
它为什么叫comet长连接的接入
我们想象的是这个模块它可能是部署在全球各地
或者是全国各地的一个节点上
然后散布在我们中国各地的这些用户
都可以接入进来
因为它处于边缘
所以它本质上也是一个边缘的节点
专门做一个场场景接入的
它核心就是做场景接接入
因为第一跳就是我们用户的网络第一跳
一般都是local ISP对本地运营商
然后连到你最近的一个节点肯定就是
比方说我在上海的话
可能连上海电信可能就会比较近
所以他可能连上海的节点优势会比较大一些
这是第一个
核心就是管理一些那种连接状态的一个模块
第二个模块叫logic叫逻辑模块
为什么会这样剥离
因为长连接模块它核心就是负责连接的
那么意味着我的客户端实际上跟我的服务端
它还有很多指令级别的一些动作
我会做一些上行的消息处理以后
也会做一些下行的消息的处理
它有很多的一些指令
是如果你把这些逻辑耦合在控制层的话
它不太容易扩展
就是这些逻辑
你如果在连接管理层的话
连接层实际上它是带状态的
因为连接它是如果你服务下线的话连接会断开
可能用户要重连
所以它实际上是带状态的
那么逻辑如果有耦合在里面
其实不利于扩展
所以我当时就把逻辑相关的东西抽象出来
comet跟逻辑层它是通过一个rpc去连接的
这样的话还有个什么好处呢
意味着几十万的连接连到comet时候
但是我comet跟我的逻辑层其实只需要建一个gRPC的连接
那么他把这种连接的一些会话信息托管在上游
托管的comet
但是业务逻辑层其实只需要一个单连接
一个单连接到我们的逻辑模块
除了用户自定义的一些这种事件
其实我们内部还有一个叫connect和disconnect的事件
就是连接和断开连接的一个事件
就是我有一个人上线了
有个人下线了
然后可以自定义的一些鉴权
鉴权接口是自定义的
然后可以存储session的一些信息
包括用户设备信息
然后他是接入至哪一个服务的
然后用户的id等等
那么最终我们业务可以通过设备 用户 还有房间room
或者是全局广播来进行推送
说白了就是可以私聊
可以匿名聊天
可以针对用户聊天
可以针对房间去聊天
然后也可以针对全局广播
就是我们所谓的以前玩游戏的那种大喇叭
是整个游戏服务里面都可以发那种会话的这种做法
当然它也可以扩展就是不同的指令
做一些不同的业务逻辑处理
logic这个模块可能跟每个公司业务都有息息相关
那么它可以完全把一些逻辑就是下放到这个里面去做
那么我业务的一个高速的迭代和发布的话
它其实不影响我们的连接模块要去重新发版
也不需要搞一些什么脚本去做一些什么脚本逻辑
做到comet里面
我们完全把它做成一个服务了
这是第二个
那么第三个叫做job服务
后台的任务
其实当时想法就是说
假设我们像1月1号发新年快乐
或者是我们到春节的时候
我们要给我们的亲朋好友发祝福的时候
那么这个时候很容易遇到消息的一个峰值
大量的人发很多消息
这些消息如果我是直接通过逻辑模块进行一些逻辑判定以后转发到comet
因为是这样的
就是你可能从a comet接入进来
但是你的用户你要私发的那个人在b comet的
那么这样的话
它其实通过逻辑模块做了一个消息中转
就从a进来的
你要发一个消息给他
那么他通过逻辑转发到b 这样的话
如果是在这种节假日或者是消息峰值的时候
特别容易产生什么压力过大
有可能就会出现事故
所以我们引入消息队列
所以我们把消息的推送先入队
先入队消息队列
然后使用job来异步消费它
我们把这种路由逻辑算好之后
job反推给我们的comet
所以我们看它的流量是这样的
这边先push到kafka
kafka这边被我们的job订阅
然后再反过来push我们的comet
最终把这个消息给路由回去
logic也好 job也好
这些模块核心还是使用的gRPC来进行通讯的
然后当然我们还依赖一些其他的外部设施
比方说像服务发现这些我就没有全部画出来了
然后我们的逻辑模块
因为要管理用户的会话信息
就是所谓的在线状态
我们当时使用的是session
当时使用的就是redis存储这些session
所以你看我们这里有get set
可以读取online一个redis的服务
除了了解它的整体架构以外
我们再来看一下它的协议是怎么设计的
为什么会出现一个协议设计
如果是我现在再重新设计的话
我可能会直接使用gRPC来用到公网上
当时是使用一个私有协议
当时在想像QQ或者微信这种
他会怎么去设计一个协议
如何方便扩展
也找了一些资料
后来也是出于一些简单的考虑
ahv设计的比较简单我们先看一下
很多人可能会问 TC PEA烟包什么的这种怎么处理
你可以想象一下你做网络编程的时候
你收到的这些一个完整的包
那么假设你收到这个包可能是不完整的
因为1个完整的报告可能是100个字节
但是如果你只收到90个字节
理论上你要等剩下那10个字节
到达之后你才能继续处理对吧
我如何知道是有100个字节
一般来说我们在做自定义的私有协议设计的时候
都会有一个叫package NES
或者Package Size
我要知道包整个这一个包一条指令
或者叫一条消息它的长度是多少
所以我们会有个Package Length
我们看到Package Length等于4个字节
这4个字节是要考虑大小端的
如果你不知道大小端的话
你可以先百度搜一下就网络大小端
这个叫Package Length是有大小端的
比方说它是Big-endian
还是Little-endian
它是一个什么样的端的
它是大端还是小端
一般除了包头还有可能还有个header
我们一般设计协议体
它会有一个header和一个body
header就是你协议体的一些这种原生头
你的body的肯定是你自定义的一些头
所以我们这个地方又有一个Header Length
就是我头有多长的
那么是两个字节
因为一般来说header不会太长
两个字就是65535
一般来说不会特别长
有了Header Length还有个协议版本
因为我们协议还是有可能会升级
比方说v1 v2 v3 对吧
可能有版本的
那么不同的Protocol Version
它的header可能内容是不一样的
所以你要是通过版本号来区分
所以前三个字段基本上是固定的
从Protocol Version往下
它可能就会有各种变化
是版本协议版本号
因为你识别到它不同的协议版本号
你的解析方式是有区别的
我们当时定的v1的版本可能就相对比较简单
还有一个指令叫Operation操作码
我这个message 我这个消息 这个指令
它对应的一个操作码
比方说它是心跳
它是发送消息
还是说做一种某一类业务逻辑
那么你只要扩展Operation 4个字就足够用了
其实有非常大
有20多亿40多亿
其实非常多的
这个是操作码
还有一个我们叫Sequence请求序号
这个id跟我们之前讲的net/rpc和HTTP2.0的二进制帧的frame id
应该是一样的
应该是一个意思
就是说我可以用一个连接发送多个上行消息
在我的comet收到这个指令之后
交给我的logic可去处理
处理完之后
他的sequence id并不会变化
我就知道是谁在client发起了一个调用
我就可以找到那个人
然后让他去换把它换行
就跟我们的net/rpc其实是一样的
sequence id是单连接多路复用的核心的一个点
除此之外还有body
前面这个部分从Package Length到Sequence id
就是所谓的我们叫goim的header
这个协议的header
那么它的body其实就是把Package Length减去Header Length
就是把减去它就得到我的包长度
那包长度我就不管你具体是什么内容
也就是说我不会去假设它的内容是什么
不会做序列化
有可能是protobuf 有可能是json
有可能是等等其他的
那么我如何识别
其实一般来说是通过协议版本号加上Operation
你就自然而然知道它的scheme应该是什么
比方说v1版本的0 operation指令
应该在我服务端注册mapping schema是长什么样的
那么我就用schema的json或者是protobuf去把它反序列化
还原成我们Go里面的内存结构
Ok这个是它的一个设计
当然我们内置的几种协议设计就包含像我们的鉴权
因为我们一般长连接建立的时候都要握手
握手其实就是做身份验证
我得知道你这个人是否是我的用户
这是第一
第二就是心跳
心跳主要是用于什么
因为长连接建立以后
如果它长期不活跃
有可能会断连
断连的原因我后面会讲
所以它需要一个心跳指令的定期保活
就是保证我的连接是通畅是ok的
那么第三个指令叫message
当时主要是叫发送消息的一个
或者叫收发消息的一个message指令
sequence的话其实就是按照请求响应来递增id就好了
你要从client维度去保证它是唯一的
我们再来看一下它的comet
刚刚说的它是一个长连接的节点
那么它通常意味着部署的距离是离用户比较近的
它跟cdn其实是非常类似非常像的
你是通过TCP或者是websocket都可以进行连接
那么通过我们应用层的heartbeat
我们知道TCP keep alive
TCP keep alive可以搜一下
那么这个其实是属于协议层
就是属于传输层的一个保活
传输层的keep alive一般是由内核来控制发送频率的
比方说我一个TCP长连接多少时间内
比方70秒80秒
没有任何的网络包
那么可能会内核会发起一个icmp的探针
来看一下对方是否是active
是否是活跃的
这么是一种内核的TCP keep alive的机制
但是这样的话其实不太方便我们自定义它的一个心跳策略
所以我们通常都是建立TCP以后
我们会定义一个heartbeat
或者叫pingpong的一个协议包来保活
保证我们的连接可用性
那节点之间
比方说我们的接入节点和我们的核心机房之间
一般是会通过云VPC来专线通讯的
那么comet北京和上海之间
我们建VPN也好或者接专线也好 是不会有的
但是它跟核心机房一般是要有一个专线
我们一般分布来说像华北的话就靠北京来覆盖
华中的话就是靠上海杭州武汉
应该叫华东
华中的话可能就是武汉这些
华南的话 广州深圳
可能还有华西
可能四川这些地方来覆盖
所以它其实是靠大的主干网的几个重要城市
来部署我们的边缘节点
那么国外可能香港 日本 美国 欧洲
可能也可以去部署
这样的话用户从本地ISP接入进来
它的成本是相对比较低的
它连通性会更好
因为comet其实最终理解为它是没有任何业务逻辑
它纯就是靠优化它的代码
优化代码来解决内存CPU的瓶颈之下
最大化它的连接数量
比方说100万个连接能吃多少内存
然后100万个连接在空闲的时候
只是单纯发心跳的时候要吃多少CPU
那么在广播消息的时候要吃多少CPU
所以它核心压测其实就是我的吞吐
我的消息的吞吐量是多少
那么长连接它的负载均衡是比较特殊一点的
因为我们看看
comet北京你不可能只布一个节点
一般在北京的某一个接入节点可能会布七八个节点
那么它的常见性的负载均衡跟我们7层负载均衡有一些区别
7层负载均衡一般是通过Nginx来做upstream
我只要多搭建几个Nginx来做转发就行了
但是长连接
如果我们想象一下像微信或者QQ这种几亿在线的
那么你前面都要搭一层负载均衡
是不是意味着我的成本是很高的
相当于我的机器要double
因为我的连接的负载均衡
是靠集中式的一个负载均衡来做的
那么所有连接必须命中到集中式的负载均衡
然后再转发到
也建一个长连接接到我们后面comet
肯定是不划算的
所以我们当时想法就是
说比方在北京的comet它有七八个节点
我用户来建连的时候
可能优先是按照IP归属的经纬度
或者他的LBS位置信息来去做智能调度的
选上北京机房
但是北京机房里面有好多个IP怎么做
我们当时想法是把comet的节点注册的
我们注册中心
我们的注册中心拿到这些IP信息以后
我们有一个叫balancer的服务
它是根据经纬度计算
还有我们最近comet节点的可用列表以及权重
因为comet可能在各个地方的机型是有些区别的
甚至新老机型也有区别
以及说它的连接的一些情况
综合得出一个决策
然后我们对外暴露了一个bff的接口
根据用户的请求IP
返回一个离他最近的一个接入节点的IP列表的一个权重
比方说用户在北京
那么我会返回一个北京机房的所有的IP列表
但是这个IP列表其实是经过一定的优先级排序的
可能刚刚说的根据它的一个机器的负载情况
连接数的情况
其实最常见的就是它的负载了
就是负载大的肯定是要不能新建连接
所以它要排在后位置
还有一块它的一个延迟
延迟的话
因为我们大区都选好了
根据它的位置信息选择了大区北京
但是在机房内的几个节点之间
基本上它的延迟除了负载会有一些影响
网络上的影响可能会比较小
如果选出来的8个结点8个IP给到你
第一个连不上
你可以做一个简单的fail over
就是直接第一个IP不行就第二个
第二个不行就第三个
通常来说在移动端会优先接入上一次成功的IP
当第一个失败的时候我们再重新计算
另外就是说你负载均衡每次要选路选到一个接入节点
怎么实际触发
在端上可以做很多策略
比方说你可以在WiFi变更的时候
你可以在网络切换的时候
你可以再冷起热起的时候
都可以在后台默默出发这个动作
然后实时更新它的
不能叫实时
应该叫准实时
获取到一个最佳的接入的节点是什么
可能有几个候选的话
那么也可以使用ping
我在我的APP端可以定期的去发一个这种探测包
然后看看它的延迟是多少
类似它的TTL是多少
我优先选这种网络质量偏高的
当然就是我们在做决策的时候
其实也要考虑到整个后面的机房
比方北京整个8个节点负载都很高的情况下
可能在做决策的时候就要可能退避到其他的机房
这个是非常要注意的
就是这个策略我没有展开讲
其实业界也有很多cdn的一些流量调度
或者机房内的一些负载均衡调度的一些paper
我最近在看的话应该就是Facebook的一个叫太极的一套系统
它其实就是管理全球的traffic一个流量的调度
我们再来看一下心跳保活的一个机制
为什么会有这个东西出现
其实做过TCP编程的同学应该都知道
需要做应用级的一个保活
其实也是面试过程中很多做TCP长连接的人都会问到的一个问题
长连接断开如下几个原因
第一
你长年接所在的进程被杀掉了
比方说我们的APP
我们的iOS推到后台以后
可能能存活个几秒钟
可能就会被我们的操作系统给杀掉
你这时候连接肯定就断开了
第二就是我们的NAT超时
我们很多这种办公网或者是一些小运营商
可能它就有NAT出口吗
我们的外网IP都是有限的
一般都是通过内网去做了一个NAT地址转化
NAT一般网关都有超时时间
比方说30秒没有任何心跳包
我NAT地址会回收
然后你可能连接就断开了
因为你一段时间内没有任何的心跳
所以我的NAT就超时了
还有第三种就是很常见的网络状态变化
比方说移动网络到WiFi的切换
断开重连
以及比方说我进到地铁里面突然进隧道了
然后没信号了
等我出来之后又有信号等等
这些状态的变化都可能导致我们的
连接会断掉
有可能要重连
还有其他的一些不可抗拒的因素
DHCP的续租
甚至说你信号很差
都是长连接断开的原因
我们再来看第二个
我们怎么去解决他这个问题
首先进程保活
进程保活指的是我的APP端尽可能不被杀掉
可能移动端的很多同学可能会比较擅长一些
可能微信这种你看他随时都能收到push
它其实可能跟小米的手机的这种有些合作
或者说跟苹果有些合作
让我的整个进程保护时间变得更长
因为im来说
它对消息的时效性要求还是蛮高的
第二个其实叫心跳保活
就是阻止NAT超时这种
所以我们需要在goim内部里面
它的client会定期它会定期发包过来
然后服务端他就收到这个包之后
它就会把他的之前的socket的超时间会续租
简单来说比方说我的socket的超时时间是30秒
那么在30秒内没有任何的这种心跳包到达
那么30秒后我会超时我就会close连接
但是如果在30秒内有一个心跳包到达
我就会收到心跳包
然后我也会回一个答复包
那么收到答复包
我们的客户的连接也会续租
也就是把它会重新调setWithTimeout
然后把它时间又改成30秒
同样的服务端也是会重新去设置长时间的
这是第二个
第三个其实是断线重连
断线重连指的是我们在一些弱网无网
或者说WiFi不稳定的情况下
连接可能很容易断开
所以需要一个策略
就是在断开以后
它需要有重连机制
然后找一个时机
比方说冷起或者一定时间内
我们监听手机的一个网络变化
它从无网变成有网或者是突然有WiFi
都可以去触发策略进行重连
还有我们的心跳时间
刚刚说的应用心跳主要是为了保活
这个心跳到底是多久
是15秒钟一次还是30秒一次
这里面我们早期其实是定时心跳
可能5分钟一次
这种策略其实不够灵活
我们看一下什么叫自适应的一个心跳时间
那么心跳可选确区间
我可以从1分钟到5分钟之间任意一个时间
我们都觉得ok的
第二就是说心跳有一个步长的增加
为什么会引入步长的机制
它其实跟我们TCP传输有点像
TCP传输的过程中
如果网络是顺畅的
每次ACK回来这个数据我没有任何丢包
那么每次我们的滑动窗口
我们TCP的滑动窗口就会变大
那么一旦丢包的话我的窗口就会瞬间缩小
因为出现丢包意味着网络可能有瓶颈
我这个时候让我的窗口变小
少发一些数据
其实也得到一个消息的抑制或者说流量的抑制
那么它的这里的策略我们引入的是什么呢
就是我当前如果有上一次是成功的心跳
那么我就延期
从60秒就可以延到30秒
那么如果失败的话就减掉30秒
为什么要这样的一个策略呢
因为我的连接如果一直成功
意味着我的网络状态良好
我的步长不需要那么高密集
如果的网络变差了
每一次都失败
那就意味着我的步长
我的重连时间
或者我的心跳要尽可能保证的短
比方说60秒
这样的话可以提前感知到是不是连接有问题
我就可以进行重连
如果你要5分钟才发现连接断掉
你才能重连肯定是有问题的
因为通过这样的一个机制
多次心跳失败
你的新的周期也变短
肯定有什么就发现了
你连接可以进行重连
在一些网络比较差的情况下
它效果就会比较好
这个是保活
我们再来看一下用户鉴权和session信息
用户鉴权是在长连接建立成功后
第一个要发的指令包
要进行鉴权
你连上来我得知道你是谁
可能会带一些自定义的token
然后我基于这个token
可能会把token下放到我的logic模块
来请求鉴权identify的服务
我鉴权成功以后我这边就会返回
你带token我就会反回用户id
最终在我comet这边我就可以维护起来
哪个接入的comet的节点
它的设备id是什么
它的token是什么
是它的rpc
就是鉴权的时候的参数
调到他就会把token请求identified这个模块
其实间接就是把这个设备id
以及它的接入的cometid
这个comet id是什么意思
因为你做消息私聊的时候
我得知道我rpc call过去的时候
我得知道call的哪个节点
所以说我得维护这个设备id
它是从哪个comet节点接入进来的
所以它要维护这样的一个映射关系
我们也叫它是一个路由的信息
我这comet有好多个节点
得知道它的IP
简单的说这个comet id你可以认为是IP端口
然后从用户维度
我得知道这个用户id它是从哪个IP端口的comet服务接入进来的
以及用户id的设备信息是什么
有了这一系列的信息以后
我其实就是把它的整个路由信息
或者叫会话信息可以存储起来的
这个地方我们看logic 它其实是依赖redis模块的
所以我们看设备id还有用户的鉴权token
这个token我是不管你的
你需要自己是实现logic的逻辑模块的一个接口
叫Auth的接口
把它换成用户id
那么comet id也是刚刚说的
就是连接所在comet的节点的IP端口信息
当然我们可以用一个id来表示
每一个comet你可以给他分配一个id
其实就是要把它的映射信息给彻底干掉
所以我们最终就生成了
刚刚说的就是连接维度和用户维度的这些信息
你在做私聊
然后再群发广播可能都会需要
我们再来看一下comet节点的它的一些内部的机制
首先我们看看协议层
协议层就是它会负责TCP和websocket的协议监听
Packet当时我们是做了一个抽象的在goim里面
Packet其实意味着每一个长连接的一个message
或者叫一个消息的包
每个包如果是特定的指令的话
它是有固定长度的
然后我们把channel可以想象的是一个移动端设备和comet之间的消息通道
消息通道相当于每个连接的一个抽象
那么channel的自然而然就可以做消息的一个收发
那么Bucket是什么呢
Bucket实际上是通过DeviceID
就是设备id维度来进行管理的
用来拆锁用的
我们为什么实现上会加上一个Bucket
如果你是用一个超大的maps来管理海量的连接
那么你肯定是要一把锁并行的去增加或者是删除
但你这样的话很容易产生一个锁冲突
我们其实是分片的
把它分成了好多个Bucket
这样的话每一个Bucket
这个Map它就可以有一把锁来维护了
我们看一下这个代码
比方说我们Bucket的对象
我们看这个Bucket的对象
它就有个自己的读写锁
它有自己读写锁
然后有一个内部有个map
map其实就是把比方有1万个连接
把1万个连接拆到了很多个Bucket里面
按Bucket来sharding的
然后里面有个它的通道的信息
通道信息订阅的就是map的string
它可能是用户ID或者是设备ID
同样bucket除了连接信息以外
它其实还放了一些房间的信息
也可以想象一下是不是说一个房间里面有很多人
比方说一个超大的直播房间里面有几十万人
我是不是也可以按bucket的维度彻底打散
所以也是一样的
核心都是为了拆锁
所以这个是bucket的对象
还有一个就是room为什么会单独抽象
像因为room它属于一个小子集
因为我们除了私聊
然后除了全员广播
那房间其实是属于一小批特定的用户
他的一个路由信息
为什么会把路由信息维护在comet
而不是放到我们的逻辑模块或者是redis里面的
因为如果每次我们要往某一个房间里面
去发消息广播的时候
你都要去查了redis把它全捞出来
有哪些人一个个的发
我认为这个成本是非常高
你可以想象一下
如果一个直播房间里面有30万个人
你从redis里面把30万个人都捞出来
捞到他们的id然后一个差他们的路由表
分别最终打包几个消息去comet去推送
是不是代价和门槛是非常高的
首先你取redis的读写的数据量非常大
因为你要查路由表
把它还原成是哪一个comet接入进来的
然后再把这个消息广播出去成本很高
所以基于这个考虑
我们是在comet的内部
把这个房间的信息
就是在内部给实现掉了
这样的话我只需要发一次广播
告诉所有的comet节点
大家往room id等于1的
房间等于1的房间去广播消息就好了
因为room信息是直接在comet内存里面就有的
那么意味着每一个comet节点有一个这样的map
这个map意味着room id里面有哪些人在里面
我们待会来看一下
刚才我们的bucket已经看过了
我们看一下room它是一个什么样的结构
就是房间
直播房间里面的人还是蛮多的
我们来看到room信息
刚刚说了room其实就是核心就是
为了优化直播类的或者说我们聊天室的这种场景
我们不希望去reading里面就是
把房间里面的人都捞出来去这样去发送
所以我们看一下room它是怎么存储的
因为你加入房间和退出房间肯定是需要有一个锁来保护的
我们这个地方注意一下
它有个叫做channel的一个next指针
那么这样的话
我们在加入房间的时候我们可以看一下
因为channel它内部实际上它是个双链表
它下一个和前面是谁
刚刚说的channels表示的是一个连接
一个用户到我们后端的一个连接
我们再看一下为什么会这样用
我们现在put一个channel加入进去
实际上就是先加锁
这个job先不管
如果这个房间被删除这种情况
假设这个房间没被删除
那么如果它的next不等于nil
然后我又把它的前向指针指向它
然后再把它的当前channel put next指向指向它的下一个指针
也就是说白了在他们两个人之间插入channel
因为它最终是用双链表维护起来的
这样的话我不需要再去用map再去存了
那么它可以快速的通过channel之间指针的跳
通过next指针跳可以去做一个消息广播
我们举个例子
我们room有个叫push的接口
那么他接受的参数叫proto
其实就是我们的packet指令包
我们看一下也是新加锁
所然后拿到room的next指针
然后你只要它不等于nil
我就不断的去枚举整个房间的所有的消息
然后一个个的都要push
其实这样的话就完成了一个房间的消息广播
它其实是靠读写锁
然后加枚举
我们可以想象一下
room这个粒度如果有几十万人
那么这个for循环要几十万次
这个成本是非常高的
所以我们前面讲到了room它实际上是要跟bucket去结合的
它需要跟bucket去打散结合的
我们再来看一下bucket的
我们看看bucket
这个里面也可以有个room是吧
相当于把room里面30万人划成了很多bucket
把它打散到bucket的粒度了
我们看看bucket的room怎么去广播的
就是bucket的维度它怎么去广播
我们看到bucket的房间广播
首先也是接受一个参数
然后它是把这个消息体发送到一个channel
我们看有个叫routines的一个channel
这个是什么东西
我们先看一下bucket
它有routines的channel
实际上它是一个channels数组
为什么要这样做
其实也是把整个发送很密集的这种情况
打散到好多个channel里面去
说白了也是为了拆锁
因为拆内部实现也是要加锁的
但是打散以后
我们是发到一个channel
异步去发的
所以我在后台又开了
好多个roomproc
roomproc它是接收刚刚channel发送来的消息
然后它会调
它会先获取bucket的room
拿到room的对象
然后再调它的push
再room的push接口进行广播
最终你可以想象为来一条消息
然后它进入到好多个bucket
每个bucket又有个room
那么每个room广播这些消息
所以它其实是一层一层去扇出
利用我们的goroutine数量
或者是利用我们的多线程的能力去批量广播这些消息的
那么让我们整个来一条消息
全员都能看到的之间的差异变得越来越小
两个人看到这个消息不会说
一个5秒以后才看得到
那肯定不行对吧
那就是来个消息
大家基本上同一时刻基本上最快时间看到
这个还是有一些优势的
所以其实核心还是为了拆锁
这个是bucket和room的一个关系
channel的话你可以理解为
它就是实现了read和write之后
对网络connection的一个封装
那么在内部的话其实做了一些小优化
就是用了一个Ring Buffer
因为我们知道客户端发指令
就是我的一个用户发指令
其实发的数量是有限的
我完全可以用一个定长的一个Ring Buffer数组
来进行一个client的消息的处理
这样的话其实也是属于叫最大化
最大化去节约内存也好
或者说减少对象数量也好
都是有一定效果的
我们看到room 看看channel的一个定义
这个channel就是表示的一个用户到一个comet连接的一个对象
首先我们这里有一个room的指针
因为我得知道这个人可能在哪个房间
所以它有个快捷指针可以直接找到
当然你也可以通过room里面再去查channel的话
相对来说会复杂一些
因为你不知道这个用户到底在哪一个房间
然后这个就是我们刚刚说的client的客户端上行指令的
一个Ring Buffer的一个缓冲区
然后这边就是bufio的reader和writer
还有我们刚刚说的next和prev的两两个指针
它是一个双链表的
那么在我们的房间
其实就是一个用户挨着一个用户去挨个发送
这样的话就是构成了comet的内部的一些实现
就是私聊群聊和广播
其实都是类似的一个思路
那么comet的内部内存其实也做了很多优化
我和举几个点
比如说像我们刚才说的Ring Buffer
你在实现上其实你可以申请一块连续内存
就是一个定长的数组
一个切片slice对吧
你可以循环使用就好了
最终对象数量是1
那么还有一些情况就是说
我们应该也涉及到bufio
你的reader writer可能都需要一个4k或者32k的一个buffer
如果一个连接有两个对象的话
假设有100万连接
你就有200万个对象
对象数量会非常多
那么GC的压力就会非常大
所以我们可能是自己做一个内存池
自己做内存池
其实也是申请一个超大的内存buffer
然后按一定大小去切割
但是这样的话其实内存是定长的
只有一个对象
你这样切割切32k可以拿出去
然后用完之后再放回来
虽然内存无法释放
但是这样对象数量非常少
所以我们在buffer层面也是做了一些优化的
所以我们引入了一个bytes buffer的对象
是我们自己实现的
它内部其实可以理解为一个定长的一个数组
一个bytes数组超大的
比方一个g
然后加上一个叫free指针
就是我们有个对象用完了
就放到一个链表里面把它放回去
你要用的时候再从列表里面先
看看有没有空闲的把它捞出来
其实非常简单的一个对象复用的做法
然后我们还有在代码实现上
为了节约内存
我们尽量会注意
让我们的对象的分配尽量在栈上
因为栈上的话
每次我的函数跑完
可能就销毁掉了
那也不需要去GC
压力也会比较小
我觉得主要还是这几块优化
另外就是我们尽可能去减少memory copy
就不要让一个消息拷来拷去
然后搞几份double
其实也是会浪费内存
因为comet它承载的连接数量非常多
所以它对性能要求其实也是蛮高的
除此之外它模块层面也有一些优化
我觉得有几个
首先是拆锁
这种全局锁
无论是连接的管理
房间的连接管理
还是用户的连接管理
因为我们有时候也要发广播消息
就是全员广播全连接的管理也好
还是房间内的连接管理也好
还有甚至别的一些对象也好
你在用的时候一定要搞清楚是不是锁冲突锁征用非常多
不是所有情况都要拆锁
一定要搞清楚你的场景是不是有好多人并发来去拿这把锁
如果有这种情况
还是要去做全局打散的
我们这个地方其实属于一种分层锁
或者说也属于一个多粒度的一个拆锁
就相当于每人用一块
这是第一个
第二个就是说并发数是要进行控制的
不要说我要一步去做一个事情的时候就立马开个goroutine
应该是在第三周讲那个并发的那块的时候
并发编程的时候就讲过
我们不要随随便便去开个goroutine
如果可以的话
你可以开固定长度的叫fixe size的goroutine
就比方你有一个worker
worker的goroutine大概四五个
利用channel去唤醒它们
让它们去工作
我觉得这个也是非常好的
那么在我们后面的代码里面其实是大量使用的
大家自己去挖源码
就像bucket room的这种广播
基本上都是用channel加固定size的goroutine去处理掉的
然后再看到我们的逻辑模块
逻辑模块的抽象我们之前也讲了
就是为了让业务跟我们的连接管理是分开的
所以我们其实从整个全局来看
它其实分为几块
首先就是我们SDK
负责做连接建立的
到了我们的goim内部以后
首先就是接入到comet这个接入节点
它负责连接管理
它提供消息的长连接的能力
我们的逻辑模块它其实属于backend的模块
backend其实就是处理什么用户服务 消息存储 群聊
你都可以把让这个逻辑模块跟它对接起来
我们最终把这些消息push到mq消费反推
所以你看通过逻辑模块做桥梁
把外围的backend的的这些服务全部给隔离出去了
你可能要做私聊
可能要做群聊
这些逻辑其实都是别的微服务
跟goim内部完全没有关系
它只是通过logic来做桥梁去call出去了
我觉得这样的话其实就非常好去扩展整套系统
最后我们再看一下job
其实做job和kafka这种机制
这种所谓的消息解藕
或者说引入消息队列来去峰的
还是非常常见的
它的坏处就是有可能会产生一定的延迟
或者说在极端情况下
消息量非常大的情况下
就会出现消息延迟的情况
这个也是一个问题
我们看一下它最终推送是怎么推送的
是我们的业务单元
比方说我现在业务单元
我的后台有个功能要给某个人发条私信
我的业务服务通过gRPC调到我的logic模块
可能是调一个叫push的接口
那么这个时候push会先查路由表
查路由表用户有没有在线
查redis有没有在线
如果在线它在哪个comet
那么它会把这个消息发送kafka
在kafka内部其实就已经告诉你这个comet
就是你应该推到哪去
那么job拿到这个消息之后
它就会找到这个comet的节点
找到具体的IP然后把这个消息发送给它
它到comet的内部以后
根据用户id可以干嘛
它是可以查到它的channel信息
拿到channel对象以后
再把这个消息发送给用户
因为我们知道刚刚bucket对象
bucket对象是有一个chs的map的
那么它有string
这个string可能是用户id
你只要根据用户的id看它命中的一个bucket
可能一般是用一致性希
或者是用普通的哈希求余都可以
找到bucket以后你再找到chs
然后查map你就可以找到channel
找到这个channel你其实就可以调push那个方法
然后把这个消息给推送出去
那么其实channel刚刚说的
它其实是net.connection的一个包装
所以其实它拿到的是最终用户的套接字
可以往对端去发送消息
所以这样的话其实就是goim的一个核心能力了
那么我们目前是支持几种
按设备推按用户推按房间推和广播
这么几种
这个其实跟goim大体的一个框架和模块我都介绍了
细节的代码我就不展开讲了
但是大家可以自己去做一些深入的阅读
然后回头再跟我们进行一些讨论
我们再介绍一下id的一些分布式的一些生成器
我就简单说一下
唯一id
无论是我们之前讲评论系统
还是我们假设接下来要做私信的一个系统
因为做私信意味着你要做消息同步对吧
比方说我不在线的时候
你给我发了好多消息
我肯定要做同步
就像我们微信没有网的时候
我登上去肯定要同步消息
这些其实都会利用到全局唯一id的做法
通常来说大家知道的就是说用uuid
uuid因为比较长又不是数字
第二就是说可能基于Snowflake的一个id生成方式
然后还有基于那种DB来维护一个步长的生成方式
还有基于数据库的增id来生成自动id的也有
还有什么基于redis DB等等
这些都是很常见的
我们内部其实像评论系统也好
还是私信系统也好
基本上核心还是用的Snowflake的id的方式
最终我们这个服务有点像什么
就是gRPC连到我们的Snowflake的一个节点
然后它会去直接返回了
当然了我们内部还有一些业务
它有些别的要求
有一些要求就是说是要某固定的一些号段的
从哪个开始之类的
我们可能会用DB来做步长的方式
我们先看一下Snowflake的它的整个的算法还是蛮简单的
我们知道Snowflake它是64bit的
就是一个int64
所以它前高位的41个bit主要是时间戳
用毫秒为单位
毫秒为单位意味着可以表示69年
那中间的10个bit的表示machine id或者叫worker id
机器id就是说不同的机器
每个人的机器id不同的话
意味着我的发号器就不会冲突
可以想象一下你等于1 我等于2
我们发出的号永远都不相同
最后12位叫sequence number
每一台机器每一个毫秒它可以发出4096个id
因为你12个bit的乘一下刚好等于4096
这样的话其实就是一个单机的发号器非常简单
如果你要分布式的话
你命中任何一台机器
因为它的工作的机器id都不同
所以说也是ok的
也是可接受的
它叫roughly order基本上是粗略的一个有序
它不能严格意义上有序
因为你如果这个时候命中两个节点
它们的机器id不同
那么它们的高41bit时间是有一些diff的nr
不能保证一定是它们有顺序关系的
它叫roughly order的就是粗略的有序的
在我们评论系统里面或者弹幕等等
我们其实还是大量使用的
还是蛮好用的
那么还有一个叫索尼flake
这个是索尼做了一些小的优化和改造
它是把前面的41个bit的改成39个bit
39个bit它的单位用10毫秒为粒度
它的时间长就变长了
它变成174年
以前的flake只有69年
所以它是10毫秒为粒度
刚刚前面的是一毫秒为粒度
那么8bit就是我们前面是12个bit
那么它现在变成8bit
意味着每10毫秒它只能产生2的8次方个id
所以它会比Snowflake稍微少一点
某一个时间内生成的id会少一些
但是它machine id会变得比较多
是16个bit
它目前用我们IP它的网段地址的255.25
就是用它的掩码的两个低位来做机器id的
那么来保证唯一
这样的话机器数量会变得多一些
各有优劣
我觉得这个其实还好
你比方说Snowflake它可能支持的机器数量是2的12次方
对于索尼flake它的机器数量就是2的16次方
就会多一些
但是它整体思路是差不多的
如果是我的话
我会建议大家说搭建一个这样的Snowflake的一个集群
那么它每个人的机器id都不同
那么你可以问任意一台机器要一个id
它都不会冲突
其实做法还是蛮简单的
你自己做一个id发号器
也不依赖什么其他的东西
还有一种做法
我们内部当时有一些要求逐步长递增的这种分布式id
我们做法也非常简单
因为Snowflake两个id之间它不连续
我们有些场景希望连续
所以我后来又做了这样的服务
这个id-server它后端依赖一个DB
它会按照业务
然后当前最大的max id 步长等等
建立有一条的记录 说白了
那么id-server它是提供一个gRPC的接口
然后可以生成id
然后携带业务标记
因为你不同的业务接入可能是有业务标记的
那么不同业务标记的分配的Aid的增的方式也不同
然后你可以部署多个id-server
然后通过数据库进行申请id步长
并且持久化最大id
你看什么意思
比方说我当前id等于1000了
我可以调他把这个max id等于2000
然后再返回
相当于要了一批配额
然后拉到我本地
然后我用完之后再去申请一个步长
再去申请一批号段拿到我本地
然后再用
这种做法就是比较简单
但是也能保证连续
然后在故障的时候
其实你只要保证下次记录自动地加1000就可以了
有个要求就是id-server我必须命中到其中某一台
比方你一个业务的它必须弄到某一台
你可以用一致性希或者说别的一些sharding的策略
就是让它命中到一台
这个其实是一种做法
但是不是特别好
我们内部其实基本上不太用这种做法了
那么还有一个就是我也见过有人用用数据库的自增字段来做的
如果他想吞吐比较高
他可能会搭建多个MySQL
然后让每个人的offset和increment不同
步长和初始值不同
然后基于数据库来发号
我忘记哪个公司也用过这种
但是我觉得强依赖数据库的做法不是特别好
还有像美团还有很多公司其实都讲过类似的id发号器的一些文章
大家也可以自己关注一下
其实有了唯一id的一个发号之后
作为这样的前提之后
我们后面就会去介绍一个完整的im的私信系统怎么去搭建起来的