构建消息推送系统之HTTP长连接实践

摘要: 构建一个推送系统，组成部分包括会话连接、心跳逻辑、消息接收、消息推送。在HTTP长连接推送系统实践上，我们可以利用servlet3的异步特性，再结合tomcat nio，可以为我们提供一个基础的技术支持。

前言

从Servlet3规范出来以后，利用Servlet3支持的异步特性，我们创建异步上下文asyncContext之后将它保存下来，同时不释放，那么这样就达到了长连接的目的。同时在配合tomcat nio的使用，利用Servlet3构建一个http长连接推送系统就有了支持基础，本篇文章将重点介绍基于Servlet3构建http长连接推送系统的实践。有关Servlet3异步的详细介绍可以参看《servlet3异步原理与实践》。

一、WEB网络结构及配置

1.1、网络结构

WEB网络结构.png
用户访问vip-->vip发布在lvs上-->lvs将请求转发给后端的haproxy-->haproxy再把请求代理转发给后端的nginx。vip实际路由发布在lvs上，但是vip配置属性在haproxy上（比如ACL, 域名，规则之类）
这里lvs转发给后端的haproxy，用户请求经过lvs，但是响应是haproxy直接反馈给客户端的，这也就是lvs的dr模式。

1.2、基本配置

我们知道http连接的特点就是一个request，一个response，然后关闭连接。这个过程包括建立连接和关闭连接。再往深处说就是调用了TCP/IP协议的三次握手，TCP协议多次传输，以及关闭连接的时候四次握手。频繁的做这些操作肯定很耗费系统的资源。从HTTP1.1以后，开始支持keepalive ，比如浏览器一旦与服务器建立连接后，会保持住一段时间，也就是减少了上面的握手和传输的次数，在这个时间段内传输数据都是复用同一个连接。当客户端主动告知关闭，或者达到了TCP关闭的条件，TCP/IP再关闭。那么通过HTTP keepalive 机制就可以让TCP连接保持住，具体保持多长时间可以通过参数来设置，下文会有介绍。
如果要保持长连接，那么根据上图的结构，浏览器与haproxy之间保持长连接(timeout http-keep-alive),haproxy与nginx之间保持长连接,nginx与tomcat之间保持长连接。我们的web应用架构一般都是如上图所示，会包含LVS、转发、反向代理。但简单起来说就是nginx+tomcat，也就是虚线框内标识的，其实我们研发人员能接触到的也是这两层，其余由运维和网络组的同学来维护。那么我重点介绍一下nginx层的配置参数。

http {
//...
keepalive_timeout       3600s; //Nginx 默认是支持 keepalive的，是通过 keepalive_timeout 设置的，默认值是75s。它表示在长连接开启的情况下，在75s内如果没有 http 请求，则关闭长连接（其实就是关闭 tcp）
keepalive_requests      800; //此值容易被忽略，它是值在 keepalive_timeout 的时间范围内，一个长连接最大允许的请求次数，如果超过此值，也会关闭此长连接。默认值为100。
gzip                    off; //这个在1.3中叙述
//...
upstream  TEST_BACKEND {
server   192.168.1.1：8080  weight=1 max_fails=2 fail_timeout=30s;
server   192.168.1.2：8080  weight=1 max_fails=2 fail_timeout=30s;

keepalive 1000;        //此处keepalive的含义不是开启、关闭长连接的开关；也不是用来设置超时的timeout；更不是设置长连接池最大连接数;而是连接程池中最大空闲连接的数量
}

server {
listen 8080 default_server;
server_name "";

location /  {
proxy_pass http://TEST_BACKEND; 
//...

proxy_http_version 1.1;         //指定 HTTP 版本，防止 1.0 版本导致 keepalive 无效。
proxy_set_header Connection ""; //清空将客户端的一些设置，防止导致 keepalive 无效

//...
}
}
}

1.3、Transfer-Encoding: chunked

普通短连接的时候浏览器根据连接关闭的状态来写response的内容。在长连接下，一段时间内传输的内容，连接都是不关闭的。因此如果没有一种机制来告知什么节点吐出内容，浏览器就只能一直等待后面是否还有数据，则迟迟不会写response的内容。那么我们可以想到利用Content-Length在传输之前标识一个包的大小，但是对于动态输出的内容，传输之前就不太好判断Content-Length的长度。在HTTP1.1最新的规范中定义了一种传输方式，就是chunked，分块编码。请求头部加入 Transfer-Encoding: chunked 之后，就代表这个报文采用了分块编码。报文中的实体需要改为用一系列分块来传输。每个分块包含十六进制的长度值和数据，长度值独占一行，长度不包括它结尾的 CRLF（\r\n），也不包括分块数据结尾的 CRLF。最后一个分块长度值必须为 0，对应的分块数据没有内容，表示实体结束。这样在长连接下动态输出内容的时候浏览器就能够判断当前这次报文结束的位置了。
在1.2中我们留了一个gzip没有介绍，我们知道开启gzip，在文本传输的情况下，所需流量大约会降至1/4-1/3。在gzip关闭的情况下，以前长连接没有任何问题，但是如果gzip打开，长连接则会失效。这是因为整个压缩过程在内存中完成，是流式的。也就是说，Nginx 不会等文件 gzip 完成再返回响应，而是边压缩边响应，这样可以显著提高 TTFB(Time To First Byte，首字节时间，WEB 性能优化重要指标)。这样唯一的问题是，Nginx 开始返回响应时，它无法知道将要传输的文件最终有多大，也就是无法给出 Content-Length 这个响应头部。因此根据chunked传输方式原理，解决了既可压缩传输也能支持长连接方式传输了。

二、HTTP长连接系统组成结构

系统组成.png

2.1、SESSION管理

SESSION是客户端到服务端的一次会话或者说是连接会话，会话信息中保存了用户PIN、连接创建时间、这次request产生的AsyncContext上下文信息。我们会将会话信息保存到内存一份，
private Map<String, Session> sessions = new ConcurrentHashMap<String, Session>(); MAP的key为用户PIN。同时把这份HASH数据也保存到redis一份，并设置好过期时间，具体设置多久没有固定的标准，我们设置是8小时。这个在心跳逻辑中，如果没有心跳会将SESSION信息删除。

2.2、心跳

心跳的目的是判断连接客户端是否还活着，隔一段时间比如5s发一次心跳包，一般是从客户端往服务端发送心跳包，我们现在HTTP长连接是从服务端往客户端发送，当初的想法是节省客户端资源。心跳的逻辑是从当前服务器内存中轮询出所有的会话信息，在发送心跳包后如果收到错误信息则标记会失败，关闭上下文asyncContext.complete();this.asyncContext = null;同时从会话列表中删除，内存和redis中都要删除。

2.3、消息接收

消息推送系统负责消息会话的创建、保持、心跳、通知推送。另外一部分就是通过MQ接收业务变更信息，通过MQ的广播机制保证每台推送系统服务器都能够收到业务变更信息。

2.4、消息推送

利用了MQ的广播所有的服务器都会收到消息，那么推送的时候是如何找到需要哪一台服务器来负责推送任务呢，在创建会话的时候我们将用户会话信息保存到了本台服务器的内存中，那么只需要判断消息中的USERPIN是否在本机内存中即可。如果不在本机内存直接丢弃该条消息。通过MQ接收到业务信息，解析出USERPIN，再根据USERPIN找到会话，拿到asyncContext，然后将通知包发送给客户端。

2.5、消息追踪

整个消息推送链相对比较长，需要做到对每个环节的埋点和跟踪，便与后续问题的跟踪处理。在业务中是通过kafka+hbase的方式，系统中把埋点数据写到本地，由采集器将数据发送到kafka，进而消费kafka插入到hbase集群。

三、HTTP长连接系统时序调用

时序图.png
结合第二节和本节的时序图我们清楚的知道实现一个推送系统主要包含会话维护、心跳、消息接收、消息推送，这其中共涉及以下三个数据包

创建会话连接包：{"protocol":1,"time":1510210650650,"state":"registered"}
心跳包：{"protocol":0,"time":1510211080780}
发送通知包：{"protocol":2,"time":1448610190241,"cmd":110001}

接下来看下重要环节的代码实现：

3.1、创建会话(连接)

public  Session createSession(String sessionId, HttpServletRequest request, HttpServletResponse response) {
//省略代码...

try {
//省略代码...
session = new HttpStreamingSession();
session.setSessionId(sessionId);
session.setValid(true);
session.setMaxInactiveInterval(this.getMaxInactiveInterval());
session.setCreationTime(System.currentTimeMillis());
session.setLastAccessedTime(System.currentTimeMillis());
session.setSessionManager(this);

session.setConnection(createHttpConnection(session, request, response));

//省略代码...

return session;
} catch (Exception e) {
//省略代码...
} finally {
//省略代码...
}
return null;
}

public void connect(){
//省略代码...
if (isClosed()) {
PushException e = new PushException("use a closed connection " + connectionId);
this.fireError(e);
}
try {
AsyncContext ac = request.startAsync();//开启上下文
ac.setTimeout(this.asyncTimeout);
ac.addListener(new AsyncAdapter() {

/**
*
* @param asyncevent
*
**/
@Override
public void onError(AsyncEvent asyncevent) throws IOException {
session.close();
}

/**
*
* @param asyncevent
*
**/
@Override
public void onTimeout(AsyncEvent asyncevent) throws IOException {
session.close();
}
});
this.asyncContext = ac;//保存上下文

} catch (Exception e) {
this.fireError(new PushException("StartAsync exception! May be the servlet or filter is not async.", e));
} finally {
//省略代码...
}
}

3.2、心跳逻辑

public void run() {//线程循环发送
while (!this.stop) {
try {
Thread.sleep(getCheckPeriod());//停5秒
} catch (InterruptedException e) {
}

if(this.stop)
break;

//省略代码...
try {
//省略代码...

Map<String, Set<String>> result = heartbeatBroadcast(MessageProtocol.generateHeartBeat());//调用心跳方法

//省略代码...
} catch (Exception e) {
//省略代码...
_logger.error("check destination! ", e);
} finally {
//省略代码...
}
}
}

protected Map<String, Set<String>> heartbeatBroadcast(String msg) {
if(isEmpty())
return null;

Map<String, Set<String>> result = new HashMap<String, Set<String>>(2);
//省略代码...
for(Iterator<String> it = httpSessionManager.getSessionKeys().iterator(); it.hasNext(); ) {
try {
identity = it.next();
session = httpSessionManager.getSession(identity);
if(session.expire()) {//只有 session 过期后才发送心跳
_logger.info("--befor hear beat --SessionId:"+session.getSessionId());
session.getConnection().send(msg);
session.access();
//省略代码...
}
} catch (Exception e) {
//省略代码...
}
}

return result;
}

3.3、消息接收

public void onMessage(List<Message> messages) throws Exception {
if (messages == null || messages.isEmpty()) {
return;
}

for (Message message : messages) {
//省略代码...

//处理消息
}

}

3.4、消息推送

public void sendMessage(String key,String context) throws DispatchException, PushException {
​
//获取USERPIN
String userPin = mem.hget(key,SessionProtocol.SESSION_FIELD_LOCALHOST);
if(!localhostUserPin.equals(localhostRedis)){//如果消息中的USERPIN不在当前主机内存中则直接丢弃该消息，由其它主机来消费发送

return ;
}
Session session = httpSessionManager.getSession(key);
if (session == null) {
_logger.info("session " + key + " no exist!");
return;
}
try {
//省略代码...

session.getConnection().send(context);
session.access();
} catch (PushException e) {
session.close();
throw new PushException(e);
} catch (Exception e) {
session.close();
throw new PushException(e);
}
}

四、半推半拉

4.1、消息存储

消息体存储.png
消息实体保存到redis集群，根据每个UERPIN组成N个HASH结构的数据体，如上图所示数据结构。因为USERPIN的数量很大，会均匀的散落到redis集群里，大量用户访问不会造成热点问题。不过有些大用户数据量会比较大，访问频率又比较高的，可以做二次HASH。

4.2、拉取方式

消息拉取图示.png
我们在长连接中推送的是消息通知，并不是消息实体。在第三节中当浏览器收到通知后会发送一次http请求带上CMD标识，服务器接收到USERPIN+CMD标识到对应的redis集群中查询数据，返回给客户端。这也就是我们说的半推半拉方式，那么我们为什么不直接把消息实体推送过去呢？推送一个简短的通知命令字，只是告诉客户端有数据变化，那么用户很有可能是不去看的，这种情况下如果直接推送实体数据，则会浪费数据传输。其实这个类似我们的公众号，比如我们收到的是一个标题和概要。如果我不去点击则不会发生文章大量内容的数据传输。

五、系统优化

5.1、NIO

长连接推送系统的最大特点就是服务器要HOLD住大量的连接，这个时候我们首先要考虑的IO模型就是要使用基于I/O复用模型的NIO。基于事件驱动利用Selector机制使用少量的线程保持住大量的连接是NIO擅长的能力。如果你使用的是tomcat7以下版本，在Connector节点配置protocol="org.apache.coyote.http11.Http11NioProtocol"，以便启用Http11NioProtocol协议。该协议下默认最大连接数是10000，可以重新修改maxConnections的值。有关tomcat nio详细介绍请参看

《深度解读Tomcat中的NIO模型》。

5.2、参数优化

一台Linux服务器可以负载多少个连接？首先我们来看如何标识一个TCP连接？系统是通过一个四元组来识别，(src_ip,src_port,dst_ip,dst_port)即源IP、源端口、目标IP、目标端口。比如我们有一台服务192.168.0.1，开启端口80.那么所有的客户端都会连接到这台服务的80端口上面。有一种误解，就是我们常说一台机器有65536个端口，那么承载的连接数就是65536个，这个说法是极其错误的，这就混淆了源端口和访问目标端口。我们做压测的时候，利用压测客户端，这个客户端的连接数是受到端口数的限制，但是服务器上面的连接数可以达到成千上万个，一般可以达到百万（4C8G配置），至于上限是多少，需要看优化的程度。最重要的一步是修改文件句柄数量限制。

查看当前用户允许TCP打开的文件句柄最大数
ulimit -n

修改文件句柄
vim /etc/security/limits.conf

soft nofile 655350
hard nofile 655350

修改后，退出终端窗口，重新登录(不需要重启服务器)，就能看到最新的结果了。还有其他有关TCP参数的修改，请参看《一台Linux服务器可以负载多少个连接？》

六、测试

在做http长连接测试的时候，无论使用chrome还是Firefox浏览器，都因为缓存的原因测试不出长连接下通过web服务动态吐内容的效果，所以我们自己写一个client。

public class HttpConnectionTest {

public static final String URL = "http://push.test.com/async?pin=123";

public static void main(String[] args) throws Exception {

ExecutorService es = Executors.newFixedThreadPool(1);
for(int i=0;i<1;i++){
es.submit(new Runnable() {
public void run() {
String URL=URL+"&client_id="+UUID.randomUUID().toString();
connection(URL);
}
});
}
}

static void connection(String url) {

InputStream is = null;
URLConnection conn = null;
byte[] buf = new byte[1024];
try {
URL a = new URL(url);
conn = a.openConnection();
is = conn.getInputStream();
int ret = 0;
while ((ret = is.read(buf)) > 0) {
processBuf(buf, ret);
}
// close the inputstream
is.close();
} catch (IOException e) {
try {
int respCode = ((HttpURLConnection) conn).getResponseCode();
InputStream es = ((HttpURLConnection) conn).getErrorStream();
int ret = 0;
// read the response body
while ((ret = es.read(buf)) > 0) {
processBuf(buf, ret);
}
// close the errorstream
es.close();
} catch (IOException ex) {
e.printStackTrace();
}
}

}

static void processBuf(byte[] buf, int length) {
System.out.println(new String(buf, 0, length));
}
}

七、总结

在这篇文章里我们从web系统的部署结构，http1.1和nginx的配置，再到实现一个http长连接系统的组成部分，推送系统的流程时序关系，最后说到系统参数调整如何来支持海量的连接。当然实现一个类似http长连接推送系统的方式还有其他比如websocket等技术，但是长连接推送系统的组成部分基本不会变也就是会话连接、心跳逻辑、消息接收、消息存储、消息推送。那么servlet3异步+tomcat nio给我们提供了一个实现http长连接推送的基础支持与实践参考。

转载请注明作者及出处，并附上链接https://my.oschina.net/wangxindong/blog/1572383

参考资料：
http://nginx.org/en/docs/http/ngx_http_upstream_module.html#keepalive
https://www.w3.org/Protocols/rfc2616/rfc2616-sec3.html#sec3.6.1