微信号:DianrongMafia

介绍:点融黑帮——一个充满激情和梦想的技术团队,吸引了来自金融及信息科技领域的顶尖人才.我们正在用技术创新改变传统金融.

干货:如何构建一个消息推送平台

2018-07-12 17:24 点融·黄义翔
01

 背 景 

 

B/S架构下很多业务场景下我们需要服务端主动推送消息到客户端,在html5之前一般使用长轮询(除此之外还有iframe流或者Flash Socket)的方式来实现,而长轮询的方式缺点很明显,频繁交互的情况下,大量的连接被建立和释放,并且交互频率受限于两次http的请求间隔,html5开始可以使用websocket全双工的通信协议,在tomcat和jetty都有实现。


虽然在java1.4以后可以用nio包实现非阻塞的websocket通信,但是使用起来太过底层和复杂;netty封装了nio,使用了大量异步和事件驱动,封装了拆包粘包,实现了网络编程和业务分离,但即便如此,使用和优化netty还是需要深厚的网络编程知识,况且我们需要的仅仅是用户与用户之间的文本消息的推送,这里我们介绍一个基于netty实现的websocket服务端框架,socket.io屏蔽了使用netty的细节,针对聊天场景做了高度封装,并且提供了包含WEBSOCKET、POLLING等多种推送方式的支持,同时支持namespace(命名空间)、broadcast(广播)、room(房间/聊天室)、event(业务事件)、应用层ack机制。

 

02

 需 求 


构建一个统一的文本消息推送平台,可以在此之上构建业务系统,比如:客服系统、内部聊天工具、站内信、事件系统等。基础平台要做到用户管理、状态管理、文本消息发送/推送(群/点对点)、图片/文件推送、未读消息管理、历史消息、会话管理。


03

 业务架构 


restful:通用接口http服务层、提供鉴权、发送消息、历史消息、离线消息、会话等相关接口

logic:业务逻辑层、消息/会话持久化、用户状态、连接状态存储、离线消息

router:路由层、消息路由、把消息推送到用户连接的connector

connector:连接层、提供消息推送服务、维护房间、ack

 

消息的上行是走的restful,在这层可以做uniauth(内部的统一登录平台)权限验证、负载均衡等,也让connector更轻量级,依赖更少;connector是有状态的,保持着用户的连接信息,同时要记录用户和connector的映射关系,以用于router层在推送消息时路由到对应的节点,connector使用单独的物理机来运行,因为虚拟主机并不能稳定保证较高并发的长连接数量。


04 

 轻量级的消息发送 


 

一个消息推送平台最重要的是高可用性和最终一致性(消息可达),所以消息发送可以做的非常轻量,发送消息时会带上客户端生成的唯一消息id,同时客户端持久化,服务端只需要把消息写到mq即发送成功,然后logic消费mq再做异步批量的消息持久化,从可用性角度来说,发送消息并不依赖websocket,而是通过restful服务,更加高可用和可扩展。消息发送的可靠性,没有依赖于rabbitmq的事务或者publisher confirms,后面介绍专门的机制来保证。

 

05

 点对点/群消息 


消息我们定义了三种事件类型,MessageCategoryEnum


MSG:文本消息,包含图片、文件、语音、文字等子类型。

IQ:操作消息,客户端用来做某些特殊动作处理。

STATUS:状态消息,和用户上下线等状态有关。

 

消息体如下

对于点对点消息来说,发送者和接受者都是一个用户id,对于群消息来说,接受者可能是一个SESSION(会话),会话需要提前创建,会话中包含若干的用户,使用socket.io的room功能,很方便的实现一个会话,room就是SESSION的id。


1. 用户上下线的时候都要主动加入/离开和自己有关的所有room

com.corundumstudio.socketio.transport.NamespaceClient.joinRoom(String room)


2. 点对点消息推送就调用         com.corundumstudio.socketio.transport.NamespaceClient.sendEvent(String 事件类型, AckCallback<?> ack, Object... 消息(SocketIOMessage))


3. 群消息推送就调用

com.corundumstudio.socketio.BroadcastOperations.sendEvent(String 事件类型, Object 消息(SocketIOMessage ), BroadcastAckCallback<T> ack)

 

06

 消息可达保证(QoS) 

 

使用六报文的方式保证消息可达,图中演示的是三报文的消息发送阶段。

1.client-A向im-server发送一个消息请求包,即msg:R;

2.im-server在成功处理后,回复client-A一个消息响应包,即msg:A;

3.如果此时client-B在线,则im-server主动向client-B推送一个消息通知包(不在线做离线存储),即msg:N;

4.client-B向im-server发送一个ack请求包表示msg已经收到,即ack:R;

5.im-server在成功处理后,回复client-B一个ack响应包,即ack:A;

6.im-server主动向client-A发送一个ack通知包,即ack:N;

7.client-A发出消息后,超过设定的某个时间没有收到ack:N,那么我们认为消息没有推送成功(用户离线除外,用户离线,服务端会模拟一个ack:N),需要客户端重发,服务端根据客户端id幂等处理;

 

在我们系统,msg:N和ack:N是connector推送,其他报文就是走http的消息发送和ack;其实ack:N可以不需要,服务器可以保存未读消息列表,接受ack:R来判断是否要重发,但是这样可靠性稍差一些,极端情况会造成消息丢失。

 

 

07

 消息顺序性 

 

首先我们要确定以什么样的逻辑排列消息的顺序,两个人聊天,首先要保证的是同一个人消息的先后顺序,其次要保证会话内消息的顺序。服务端不能以持久化时生成自增id作为顺序,因为异步消费不能保证顺序(某些mq,如kafka以用户为key可以保证消息顺序性,但也存在重试问题)。从客户端的角度来说,可以通过同步发送保证消息的顺序性,所以服务端可以在发mq之前通过redis生成会话内自增id,推送的时候可能会存在重发和乱序,就需要客户端根据服务端生成的id做幂等和重排序。


如果不依赖于redis,可以使用服务器时间(精确到秒)+客户端生成的自增id作为排序字段,但这样就要求集群的时间同步在一秒内。

  

08

 未读(离线)消息 


关于未读消息的存储有两种方案:写扩散和读扩散,写扩散即未读消息在持久化时就针对每个用户保存一份,读扩散就是利用群消息的偏序特性,只保存用户在会话内ack的最后一条消息id,无论是哪种方式,未读消息都只是保存消息的id,未读消息的推送仅在上线或者连接的时候。


写扩散:

session_user(session_id, user_id);//会话-用户

messages(msgid,session_id,sender_user_id,time,content);//消息

user_messages(user_id, msgid, session_id);//用户-未读消息

 

1. 消息持久化后,如果接受用户不在线,那么在user_messages为每个用户新增一条记录;

2. 用户ack后,删除t_user_messages记录;

3. 用户查未读消息,先查user_messages,再查messages。

 

读扩散:

session_user(session_id, user_id, last_ack_msgid);//会话-用户

messages(msgid,session_id,sender_user_id,time,content);//消息

user_messages(user_id, msgid, session_id);//用户-未读消息

 

1.消息持久化,推送在线用户;

2.用户ack,更新last_ack_msgid;

3.用户查未读消息,先查last_ack_msgid,再查messages。

 

优化点:

1. 未读消息user_messages表或者last_ack_msgid字段的更新很频繁,可以用redis来实现存储;

2. ack的频率很高,可以让客户端批量或在一个时间段内ack,减少请求,即使消息重发,也有客户端去重;

3. 未读消息可能存在消息量很大的情况,可以设置过期时间,过期后即表示该消息已读;大量未读消息不适合做推送,最好通过http分页拉取,甚至可以把拉取请求和ack请求合并。

 

09

  调  优 

 

1. Jdk nio会判断Linux kernels >= 2.6,是则使用level-triggered epoll,否则使用select/poll,而netty实现了edge-triggered epoll,在高并发场景下性能略优,也是netty推荐的模式,设置socket.io的useLinuxNativeEpoll=true,使用linux epoll, NioEventLoopGroup 替换成EpollEventLoopGroup,NioServerSocketChannel 替换成 EpollServerSocketChannel;

2. 要建立大量连接需要修改最大文件句柄数,修改sysctl.conf的fs.file-max = 1000000,同时修改limits.conf

soft  nofile  1000000

hard  nofile  1000000

3. 设置TCP_NODELAY=true禁用nagle算法;

4. 设置全连接accept队列长度,SO_BACKLOG=65535,同时设置内核参数net.core.somaxconn=65535

5. 根据消息的平均大小,调整tcp socket读写缓冲区的默认值net.ipv4.tcp_wmem=4096 4096 4161536 net.ipv4.tcp_rmem=4096 4096 4161536

6. 调整net.ipv4.tcp_mem=786432 2097152 3145728,当连接量较大时,单个连接占用的堆外内存接近于4096+4096=8k,如果有50w连接,需要至少4g内存,那么tcp_mem最大值要大于4g;

7. 上行数据使用AdaptiveRecvByteBufAllocator动态分配ByteBuf;

8. 业务逻辑不要占用EventLoop线程,启用业务线程池处理,否则可能阻塞I/O;

9. Tcp层的心跳只能检测连接,不能确定应用可用,所以有了应用层心跳, socket.io默认25s,可以调整为180s,频率过高会占用大量带宽和流量,过低可能会导致连接断开。同时关闭tcp心跳保活SO_KEEPALIVE=false;

10. Netty的boss threads设置为1(1个监听端口),worker设置为16(cpu*2)。

 

10

  Socket.io 


1.服务端

netty-socketio是java版的服务端实现,目前只支持xhr-polling和websocket 两种transport。

NamespacesHub是持有所有命名空间的一个map,value是Namespace。

Namespace持有当前命名空间下的所有client连接room和client的对应关系。

NamespaceClient代表当前命名空间下的一个client连接。

ClientHead代表一个client连接,持有会话id、TransportState以及子NamespaceClient(一个Channel可以对应多个命名空间)。

TransportState持有一个Packet的无锁队列,以及一个netty的通道Channel,通过此向客户端推送消息。

Packet就是一个socket.io协议下的数据包。

 

socket.io为我们封装了支持websocket协议的ChannelHandler,其中就包括最重要的InPacketHandler,InPacketHandler提供了协议解码,用来处理非OPEN,UPGRADE的PacketType事件:CONNECT(连接)、PING(应用层心跳)、UPGRADE(升级websocket)、CLOSE(连接关闭)、MESSAGE(业务消息),socket.io使用自定义协议Packet,并使用json序列化的方式传输。


2.客户端

2.1 连接

var socket =  io.connect('http://127.0.0.1:9092/namespace');//发起客户端连接,客户端sdk发起connect的时候会发出下面的若干请求

 

https://crm-connector-dev.dianrong.com/socket.io/?EIO=3&transport=polling&t=MFJ3jBa

Response:{"sid":"bbfd934a-9665-4138-9593-f23fe2836215","upgrades":["websocket"],"pingInterval":180000,"pingTimeout":60000}

会生成一个sid表示本次连接会话,upgrades:websocket表示客户端可以把协议升级为websocket而不再使用轮询,而客户端也会判断当前浏览器是否支持websocket来决定是否升级

 

https://crm-connector-dev.dianrong.com/socket.io/?EIO=3&transport=polling&t=MFJ3jD0&sid=bbfd934a-9665-4138-9593-f23fe2836215

Response:[{"code":"0","message":"success"}]

客户端发送了一个业务上的注册事件给服务端,并接受了一个成功的返回,在websocket没有连接成功前,会一直使用长轮询方式接受服务端推送

 

wss://crm-connector-dev.dianrong.com/socket.io/?EIO=3&transport=websocket&sid=bbfd934a-9665-4138-9593-f23fe2836215

建立websocket长连接

 

2.2 监听

监听一个msg业务事件

socket.on(‘msg’, function(data) {

                     //do something

});

 

监听断开连接事件

socket.on('disconnect', function() {

                     //do something

});

 

 

3.Socket.io websocket协议包


每一个Frame数据帧都以数字开头,三位数字分别表示command(命令):messageType(消息类型):req_id(自增id)

第一行 2probe 表示首次ping

第二行 3probe 表示首次pong

第三行 5 表示UPGRADE,升级为websocket

第四行 421[“iq...] 表示iq事件消息

 





12

往期精彩

· 《我不是药神》| 生而为人的各种难

· 看人生赢家是怎么赢球的

· MLP技术在信用卡欺诈检测中的应用探究

· 程序员和艺术有个约会

 
点融黑帮 更多文章 何为“产品力”—— 产品经理年度必听讲座 年度升值加薪必听讲座 看人生赢家是怎么赢球的 《我不是药神》| 生而为人的各种难 程序员和艺术有个约会
猜您喜欢 ✅ 做人,千!万!不要“透支”你的信任! 域名和接入 花川学院的继承者 - 小波说雨燕 全新上线 创业起步期,应该如何做技术选型?