关于im:如何快速构建可靠的分布式IM聊天系统

起源：如何疾速构建牢靠的分布式IM聊天零碎

tlnetim 聊天我的项目是一个分布式 im demo，基于 tlnet http框架和 tldb数据库。tldb是一个高性能的分布式数据库，基于tldb能够疾速构建分布式系统。

tlnetim 的开源程序：

https://github.com/donnie4w/tlnetim
https://gitee.com/donnie4w/tlnetim

tlnetim次要的性能：多聊天室多人聊天零碎，程度扩大多服务器部署构建分布式
im.tlnet.top与im2.tlnet.top 是分布式系统的两个不同的服务

用户能够连贯任意一个服务器相互通信
除去局部存储数据的实现，im的逻辑代码理论只有几十行，基于tldb mq代码实现分布式的代码也只有几十行。

room, ok := wsmap.Get(ws)if !ok {    if wa.ATYPE == LOGIN {        if iu, ok := getUserInfo(wa.MSG); ok {            room = strings.TrimSpace(wa.ROOM)            store(ws, iu, room)            //记录登录日志            orm.Insert(&ImLog{UserId: iu.Id, Room: room, Time: TimeNow()})            ws.Send(wsack{ATYPE: wa.ATYPE, USERNAME: iu.Name, ICON: iu.Icon, TIME: TimeNow(), ROOM: room}.toJson())            immq.PubId(room, iu.Id)            //返回好友列表            if *UseRobot {                ws.Send(wsack{ATYPE: FRIEND, USERNAME: robot.Name, ICON: robot.Icon, LABEL: robot.Label}.toJson())            }            broadcastToSelf(&wsack{ATYPE: FRIEND}, ws, room)            //告诉好友            broadcast(&wsack{ATYPE: FRIEND, USERNAME: iu.Name, TIME: TimeNow(), ICON: iu.Icon}, ws, room, true, true)            //返回聊天室 最新N条数据            if id, _ := orm.SelectIdByIdx[ImMessage]("Room", room); id > 0 {                startid := id - 20                if startid < 0 {                    startid = 0                }                if ims, _ := orm.SelectByIdxLimit[ImMessage](startid, 21, "Room", room); ims != nil {                    for _, im := range ims {                        var u *ImUser                        if im.UserId > 1<<60 {                            u = robot                        } else {                            u, _ = orm.SelectById[ImUser](im.UserId)                        }                        if u != nil {                            ws.Send(wsack{ATYPE: MSG, USERNAME: u.Name, ICON: u.Icon, MSG: im.Content, TIME: im.Time}.toJson())                        }                    }                }            }        } else {            ws.Send(wsack{ATYPE: NOPASS}.toJson())        }    }} else if wa.ATYPE == MSG {    iu, _ := getIu(room, ws)    t := TimeNow()    //保留聊天信息    if _, err := orm.Insert(&ImMessage{UserId: iu.Id, Content: wa.MSG, Time: t, Room: room}); err == nil {        //发送聊天数据        broadcast(&wsack{ATYPE: MSG, USERNAME: iu.Name, MSG: wa.MSG, TIME: t, ICON: iu.Icon}, nil, room, true, false)    }}

tlnet将服务器的websocket封装为三个阶段：

关上连贯阶段：OnOpen
读到信息阶段：WS.Read()

链接敞开或出错阶段：OnError
这些封装，让websocket的用法与应用一般的http服务基本一致：

  wc = &tlnet.WebsocketConfig{}  //websocket断开时，触发OnError。删除wsmap中的连贯  wc.OnError = func(self *tlnet.Websocket) {  if r, ok := wsmap.Get(self); ok {      if u, ok := getIu(r, self); ok {          //掉线告诉          broadcast(&wsack{ATYPE: LOGOUT, USERNAME: u.Name}, nil, r, true, true)      }    }  }  //wc.OnOpen 用在连贯胜利时调用  //hc.WS.Read() 读取websocket接管的数据  var wa wsack  if err := json.Unmarshal(hc.WS.Read(), &wa); err == nil {       parse(wa, hc.WS) //解析并解决信息  }

基于tldb的MQ音讯订阅公布，简洁地实现分布式构建

  this.mq = cli.NewMqClient("ws://127.0.0.1:5000", "mymq=123")         //mq服务器地址与用户名明码  if err := this.mq.Connect(); err != nil {                           //mq.Connect() 连贯服务器  panic("mq connect err:" + err.Error())  }  this.mq.MergeOn(1)              //设置服务器信息聚合发送到客户端，1示意数据包大小下限为1MB  this.mq.Sub("immsg")            //订阅topic：immsg  this.mq.Sub("id")               //订阅 topic：id  this.mq.Sub(fmt.Sprint(nodeId)) //订阅本节点信息  //解决订阅信息，接管公布函数PubMem()发送的数据,不存储信息  this.mq.PubMemHandler(func(jmb *JMqBean) {  defer MyRecover()  var ms mqws  json.Unmarshal([]byte(jmb.Msg), &ms)  switch jmb.Topic {  case "immsg":      if ms.NodeId != nodeId {          broadcast(ms.Wa[0], nil, ms.Room, false, false)      }  case "id":      if m, ok := roomap.Get(ms.Room); ok {          wss := make([]*wsack, 0)          m.Range(func(_ *tlnet.Websocket, vu *ImUser) bool {              if ms.UserId != vu.Id {                  wss = append(wss, &wsack{ATYPE: FRIEND, USERNAME: vu.Name, ICON: vu.Icon, LABEL: vu.Label})              }              return true          })          immq.PubInfo(ms.NodeId, ms.UserId, ms.Room, wss)      }  case fmt.Sprint(nodeId):      if k, ok := wamap.Get(ms.UserId); ok {          for _, v := range ms.Wa {              k.Range(func(w *tlnet.Websocket, _ int8) bool {                  w.Send(v.toJson())                  return true              })          }      }  }  })  //解决订阅信息，这里应用json格局，接管公布函数PubJson()发送的数据，也能够应用 PubByteHandler()对应PubByte()  this.mq.PubJsonHandler(func(jmb *JMqBean) {  defer MyRecover()  var ms mqws  json.Unmarshal([]byte(jmb.Msg), &ms)  switch jmb.Topic {  case "immsg":      if ms.NodeId != nodeId {          broadcast(ms.Wa[0], nil, ms.Room, false, false)      }    }  })

告诉好友函数：

  broadcast(&wsack{ATYPE: FRIEND, USERNAME: iu.Name, TIME: TimeNow(), ICON: iu.Icon}, ws, room, true, true)

broadcast理论是tlnetim实现的一个音讯路由性能，将状态(上线，下线等)，收回信息等进行播送，其中包含路由给其余分布式节点。tlnetim在发送给其余节点中的实现采纳tldb MQ的订阅公布，次要用PubMem 与PubJson，这两个公布函数也是有区别的，PubMem 不存储公布的信息，个别用于状态信息的公布，比方登录信息，下线信息等。PubJson 公布的信息会记录在tldb中，保障信息不失落。

针对在并发量比拟大的分布式系统。tldb MQ对音讯解决提供一些解决方案

  this.mq.MergeOn(1)    //服务器信息聚合发送到客户端，1示意数据包大小下限为1MB  this.mq.SetZlib(true) //服务器消息压缩

MQ 提供音讯的可靠性保障：客户端信息回执，拉取信息等：

  this.mq.RecvAckOn(10) //音讯回执与重发工夫的设置，10示意如果客户端10内不回执服务器信息，则服务器会再次发送信息  this.mq.PullJsonSync("immsg",1) //拉取topic：immsg id为1的信息 json格局  this.mq.PullByteSync("immsg",1) //拉取topic：immsg id为1的信息 二进制格局  this.mq.PullIdSync("immsg") //拉取 topic：immsg的最大信息id

MQ反对各个客户端对 MergeOn ，SetZlib， RecvAckOn这些性能依据不同理论状况各自设定。

SetZlib 是用zlib压缩音讯发送，在大量应用zlib压缩时，服务器会耗费大量内存，所以并非每个音讯都适宜压缩发送，应该依据理论状况，如果音讯体比拟大，或采纳聚合信息，也就是多少个音讯聚合发送，总数据比拟大，这时压缩信息会有比拟好的成果，压缩比例较大，达到减小传输工夫，进步吞掉量的成果。相同，如果音讯体自身曾经较小，压缩成果不佳，这时压缩音讯反而减少服务器压力。

RecvAckOn 是音讯回执，保障音讯不失落。MQ服务器给节点推送信息时，节点会发Ack给MQ告知音讯已收到，否则，当MQ服务器没有收到Ack时，会一直给节点推送信息；相似这样的性能，不论是tldb MQ还是其余MQ，节点的状态应该实时监控，如果节点压力过大，比方，某个服务节点cpu被打满了，此时，节点可能无奈回复ack给mq服务器，mq服务器因为收不到回执，会呈现大量信息积压。

不开启RecvAckOn的状况下，tldb mq也提供了反对客户端信息不失落的办法。如果不开启 RecvAckOn，对于同一信息，mq服务器只给节点发送一次。如果节点狐疑服务器信息没有达到，即信息丢了(须要客户端依据理论业务状况实现一个狐疑发现策略)，能够通过拉取id函数PullIdSync拉取订阅主题的最大id，与本地的id比拟，来判断本地是否有信息未读取到，通过拉取信息函数PullJsonSync或PullByteSync 将mq服务器的信息拉取到本地。

即时通讯零碎，依据不同业务要求，有不同的实现。tlnetim聊天IM在协定方面只是简略设计，只是demo，不实用简单或残缺的IM零碎。残缺的IM聊天协定能够参考xmpp即时通讯协定。

有任何问题或倡议请Email：donnie4w@gmail.com或 http://tlnet.top/contact 发信给我，谢谢！