goim 工作流程

goim 工作流程  其中 comet服务 在最前面对外提供服务，提供 tcp 和 websocket 两种服务。 接受到消息之后 交给 logic 模块处理。logic 模块在调用 router 查询路由信息。将发送的消息交给 kafka，再由 job 服务读取kafka的消息，推给 comet服务。 拆分成这么多模块的好处是可以多实例部署。其中 在 comet，router，logic 模块都提供了性能监控 net/http/pprof 同时还提供了monitor 服务，服务启动之后访问 xxx:port/monitor/ping 返回 ok 说明服务启动正常，方便运维检查系统。 **各模块功能分析** **comet 模块** comet 模块是在最前端，主要负责和client的链接保持，同时接受，发送消息，通知到客户端。检查链接是否断开。 comet 属于接入层，非常容易扩展，直接开启多个comet节点，修改配置文件中的base节点下的server.id修改成不同值（注意一定要保证不同的comet进程值唯一），前端接入可以使用LVS 或者 DNS来转发。 **logic 模块** logic 模块是comet 模块调用的，接受 comet 模块的命令， 然后进行处理，再发送的消息的kafka队列上，同时链接 router 模块，记录用户的 uid server room 等信息。同时获得router模块的信息。 logic 属于无状态的逻辑层，可以随意增加节点，使用nginx upstream来扩展http接口，内部rpc部分，可以使用LVS四层转发。 **router 模块** router 主要记录了用户的session信息。存储在 Bucket 对象里面。 **job 模块** job服务每秒都在同步 comet的信息，然后再读取 kafka队列的信息。 push 到相关的comet服务器上。用户就接受到了消息。 job 根据kafka的partition来扩展多job工作方式，具体可以参考下kafka的partition负载。 **新添加功能** 1、用户通过websocket 发消息，comet模块使用 rpc 调用logic 将消息发送到kafka，job消费kafka的消息推送给用户。 2、将 kafka 消息队列改成 nsq 消息队列。kafka 与 nsq 对比 kafka消息会固化，存文件，nsq默认是不保存的 kafka消息因为固化下来，所以是保序的，nsq传递时候通常是无序的，当然你也可以保留下信息去check时间戳，因此nsq更适合处理数据量大但是彼此间没有顺序关系的消息。 nsq支持延时消息的投递，比如我想这条消息5分钟之后才被投递出去被客户端消费，较于普通的消息投递，多了个毫秒数。延时消息可用于以下场景，比如一个订单超过30分钟未付款，修改其状态 或者给客户发短信提醒。 3、router挂掉后如何处理？ 完成把客户端的心跳消息，也通过logic，都发到router。如果router挂掉，可以根据这些心跳重建在线用户map。 心跳消息发给router，还有一个好处是，能清理太久没有下线的用户脏数据。  **op状态** 握手：op = 0 握手返回： op = 1 心跳：op = 2 心跳返回： op = 3 发消息：op = 4 消息返回：op = 5 授权用户： op = 7 返回： op = 8 OP_PROTO_READY // 消息推送 OP_PROTO_FINISH // 关闭退出 websocket.go **客户端链接逻辑** ~~~ go acceptWebsocket 监听客户端链接 go serveWebsocket(server, conn, r) lAddr := conn.LocalAddr().String() 获取服务端ip rAddr := conn.RemoteAddr().String() 获取客户端ip server.serveWebsocket(conn, rp, wp, tr) ch.Reader.ResetBuffer(conn, rb.Bytes()) 读缓冲，重置IO，可读游标重置为0，且b.buf变为buf tr.Add 添加一个时钟对象 websocket.ReadRequest(rr) websocket，读取请求 wp.Get() 获取一个写缓冲区 ch.Writer.ResetBuffer(conn, wb.Bytes()) 重置IO，可写游标重置为0，句柄变为w，缓冲区变为buf websocket.Upgrade(conn, rr, wr, req) 交换协议 ch.CliProto.Set() 获取一个proto对象的引用，用于写，不会移动可写游标 key, roomId, hb, err = server.authWebsocket(ws, p) 握手->注册，返回 subkey,roomId,heartbeat 握手->注册逻辑 server.operator.Connect(p); PRC -> 调用 logic RPC.Connect 通过 token 获取 用户 userId、roomId rpc -> 调用 RouterRPC.Put 通过 UserId、ServerId、RoomId 存 session server.Bucket(key) 使用subkey根据CityHash32算法，分配到对应的Bucket b.Put(key, roomId, ch) 分配到chs和rooms tr.Set(trd, hb) 设置更新定时器数据。 DefaultWhitelist.Contains(key) 判断是否是白名单用户 server.Stat.IncrWsOnline() 增加 websocket online 统计 ~~~ **客户端链接逻辑流程图**  **收发消息逻辑** ~~~ 参数： conn net.Conn 链接 rp, wp *bytes.Pool 读写缓冲池 Pool内存组织如下，Pool是一个链式存储的栈，数据从栈顶出，同时数据也从栈顶回收。 tr *itime.Timer 最小堆算法实现的时钟对象 协程 go server.serveWebsocket() 处理读消息（接收心跳消息、接收用户发消息） for { ch.CliProto.Set() 从 Ring 上，获取一个proto对象的引用，用于写，不会移动可写游标 p.ReadWebsocket(ws) Read Websocket 读websocket发来的消息 判断是 心跳检测 or 发消息操作 p.Operation == define.OP_HEARTBEAT 心跳检测 p.Operation = define.OP_HEARTBEAT_REPLY 心跳回应 else 发消息操作 err = server.operator.Operate(p) p.Operation = define.OP_SEND_SMS_REPLY 发消息回应 ch.CliProto.SetAdv() 移动 Ring 的可写游标 ch.Signal() 将消息 写入 c.signal chan *proto.Proto } 协程 go server.dispatchWebsocket() 处理写消息（心跳回应、推送消息） for { ch.Ready() 消费 c.signal chan *proto.Proto 的消息 switch p 判断消息类型（ProtoReady消息回应、ProtoFinish页面关闭、default 推送消息） ProtoReady->自己发的消息 p, err = ch.CliProto.Get() 从 Ring 上 获取一个proto对象的引用，用于读，不会移动可读游标 p.WriteWebsocket(ws) 消息推送（发消息返回结果） ch.CliProto.GetAdv() 移动可读游标 ProtoFinish页面关闭 -->走断开链接逻辑 default push推送消息 err = p.WriteWebsocket(ws) 消息推送 ws.Flush() 释放写缓冲池 } ~~~ **收发消息流程图**  **断开链接逻辑** ~~~ err = p.ReadWebsocket(ws); err != nil 浏览器断开链接 ws.ReadMessage(); err != nil err = ErrMessageClose b.Del(key) 根据subkey 删除 chs和rooms 上的Channel和room tr.Del(trd) 删除时间计数 ws.Close() 关闭websocket链接 ch.Close() 发送关闭channel信号 "proto.ProtoFinish" rp.Put(rb) 归还一个缓冲区 server.operator.Disconnect(key, roomId); 等待注销结果 注销过程 server.operator.Disconnect(key, roomId); rpc -> 调用 logic RPC.Disconnect 根据 subKey 返回UserId、seq rpc -> 调用 RouterRPC.Del 根据 UserId、Seq、RoomId 删除 session p = ch.Ready() （等待 return <-c.signal ）接受到 "proto.ProtoFinish" 检测到浏览器断开 switch p -> proto.ProtoFinish: 进行退出操作 ws.Close() 关闭链接 wp.Put(wb) 归还一个缓冲区 dispatch goroutine exit 退出协程 go server.dispatchWebsocket() 注销结果返回 server tcp goroutine exit 退出协程 go server.serveWebsocket() server.Stat.DecrWsOnline() 减少 websocket online 统计 ~~~ **断开链接流程图**  **几个重要的结构体** 做为典型代码即注释的开源项目，goim 基本无太多阅读障碍，几个逻辑点梳理下很快就会明白。 Bucket: 每个 Comet 程序拥有若干个 Bucket, 可以理解为 Session Management, 保存着当前 Comet 服务于哪些 Room 和 Channel. 长连接具体分布在哪个 Bucket 上呢？根据 SubKey 一致性 Hash 来选择。 Room: 可以理解为房间，群组或是一个 Group. 这个房间内维护 N 个 Channel, 即长连接用户。在该 Room 内广播消息，会发送给房间内的所有 Channel. Channel: 维护一个长连接用户，只能对应一个 Room. 推送的消息可以在 Room 内广播，也可以推送到指定的 Channel. Proto: 消息结构体，存放版本号，操作类型，消息序号和消息体。 **job模块消费kafka逻辑** pushRoutines []chan *proto.MPushMsgArg 多播 broadcastRoutines []chan *proto.BoardcastArg 广播 roomRoutines []chan *proto.BoardcastRoomArg 房播 ~~~ 消息从kafka集群消费，经过kafka模块转发至push模块，push模块对消息预处理/过滤/分类，然后发至不同的 comet 信道中 kafka -> push(msg.Value) 消费kafka消息 push -> push() 判断消息类型，多播、广播、房播 多播：define.KAFKA_MESSAGE_MULTI: -> go processPush(pushChs[i]) 根据启动的 go 协程取模分配 -> comet.mPushComet(serverId int32, subKeys []string, body json.RawMessage) 消费 pushChs []chan *pushArg chan 的消息 -> Push(arg *proto.MPushMsgArg) 将消息推到 pushRoutines []chan *proto.MPushMsgArg chan 中 广播：define.KAFKA_MESSAGE_BROADCAST: -> broadcast(m.Msg) 调用所有的 *Comet -> c.Broadcast(&args) 将消息推到 broadcastRoutines []chan *proto.BoardcastArg chan 中 房播：define.KAFKA_MESSAGE_BROADCAST_ROOM: -> roomBucket.Get(int32(m.RoomId)) 获取房间信息 -> NewRoom(roomId, b.round.Timer(b.roomNum), b.options) -> go r.pushproc(t, options.BatchNum, options.SignalTime, options.IdleTime) -> broadcastRoomBytes(r.id, buf.Buffer()) 根据roomid获取 *Comet -> c.BroadcastRoom(&args) 将消息推到 roomRoutines []chan *proto.BoardcastRoomArg chan 中（根据消息总数与options.RoutineSize取模） -> room.Push(0, define.OP_SEND_SMS_REPLY, m.Msg) 将消息保存到room.proto chan *proto.Proto chan 中 main.InitComet(addrs map[int32]string, options CometOptions) 初始化 comet -> go c.process(pushChan, roomChan, broadcastChan) 通过 rpc 调用 comet，作为pushChan，roomChan，broadcastChan信道的消费者 ~~~ **job模块消费kafka逻辑流程图**  **logic 模块 push http推送接口逻辑** ~~~ http接口-->kafka生产者逻辑 单人推送：push -> r.URL.Query().Get("uid") 获取userId -> ioutil.ReadAll(r.Body) 获取body -> genSubKey(userId) 通过用户ID生成subKeys；同一个用户可以同时多处登陆或者同处多实例登陆，它们都会被同等对待 -> rpc 到 router 模块 RouterRPC.Get 获取指定用户的session信息 -> mpushKafka 根据serverId、subkeys、Msg ,push 到 kafka -> OP: define.KAFKA_MESSAGE_MULTI 单播 -> producer.Input() 多人推送：Pushs -> ioutil.ReadAll(r.Body) 获取body -> parsePushsBody(body []byte) 将body转成userIds、Msg -> genSubKeys(userIds) 并行获取多个用户信息，返回值为map[comet.serverId][]subkey. -> mpushKafka 根据serverId、subkeys、Msg ,push 到 kafka -> OP: define.KAFKA_MESSAGE_MULTI 单播 -> producer.Input() 房间推送：PushRoom -> ioutil.ReadAll(r.Body) 获取body -> param.Get("rid") 获取roomId -> strconv.ParseBool(param.Get("ensure")) 获取ensure，是否强推 -> broadcastRoomKafka 根据roomId、Msg、ensure ,push 到 kafka -> OP: define.KAFKA_MESSAGE_BROADCAST_ROOM 房间内广播 -> producer.Input() 广播：PushAll -> ioutil.ReadAll(r.Body) 获取body -> mpushKafka 根据Msg ,push 到 kafka -> OP: define.KAFKA_MESSAGE_BROADCAST 广播 -> producer.Input() ~~~ **logic 模块 push http推送接口逻辑流程图**