使用jetty的continuations实现"服务器推"

ouzhoudijie

　　在实际的开发中,我们可能会有这样的场景:许多客户端都连接到服务器端,当有某个客户端的消息的时候,服务器端会主动"推"消息给客户端,手机app的推送是一个典型的场景(IOS的推送都是要经过苹果的服务器的,一般是通过苹果的APNS服务来实现,不需要做过多的开发,安卓的推送就需要我们自己来实现了)
　　我们可选的技术方案实际上是很多的,使用netty这样的异步的网络通信框架或者servlet容器提供的异步的方案都是可以实现的,它们的理念都是一样的,异步和事件驱动,客户端请求服务器,当服务器没有需要推送的数据(或者是需要执行很长时间的IO操作)的时候,请求会被挂起,当服务器端的数据准备好的时候(例如需要向客户端推送一个消息的时候,或者是服务器端IO操作执行完毕了)请求会被重新激活,数据返回客户端.
　　使用jetty的continuations或者是netty来实现这两种是我觉得比较好的实现方案,今天介绍一下如何使用jetty的continuations来实现一个服务器推的原型,和正式环境中向安卓手机的推送的实现方法是完全一样的
　　continuations介绍:jetty的continuations是jetty实现的实现异步请求和事件驱动的组件,从jetty7起,continuations不止在jetty中可以使用,任何支持servlet3.0规范的servlet容器都可以使用continuations来实现异步和事件驱动,相比servlet3.0规范中的异步servlet,continuations提供了更加简化的编程模型.
　　目标:用浏览器请求服务器的一个URL(用浏览器来模拟我们的客户端),实现任何时候当服务器需要推送数据的时候,浏览器能够立即显示出来
　　我们需要提供两个接口:提供给客户端做长连接的接口,向客户端发送数据的接口
　　提供给客户端连接的servlet:



package com.jiaoyiping.websample.asyncServlet.jetty;
/*
* Created with Intellij IDEA
* USER: 焦一平
* Mail: jiaoyiping@gmail.com
* Date: 2016/10/23
* Time: 23:52
* To change this template use File | Settings | Editor | File and Code Templates
*/
import org.eclipse.jetty.continuation.Continuation;
import org.eclipse.jetty.continuation.ContinuationListener;
import org.eclipse.jetty.continuation.ContinuationSupport;
import javax.servlet.ServletException;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.Writer;
import java.util.Map;
@WebServlet(urlPatterns = "/pull", asyncSupported = true)
public class ContinuationServlet extends HttpServlet {
@Override
protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
String user = req.getParameter("user");
Map<String, PushAgent> pushAgentMap = (Map<String, PushAgent>) req.getServletContext().getAttribute("agentmap");
if (pushAgentMap.containsKey(user)) {
PushAgent pushAgent = pushAgentMap.get(user);
Continuation continuation = ContinuationSupport.getContinuation(req);
continuation.setTimeout(90000000);
//第一次请求进来
if (continuation.isInitial()) {
resp.setContentType("text/evf;charset=utf-8");
resp.setHeader("Connection", "keep-alive");
resp.setHeader("Keep-Alive", "timeout=2000");
PushAdapter pushAdapter = new PushAdapter(continuation, pushAgent);
continuation.setAttribute("adapter", pushAdapter);
continuation.addContinuationListener(new ContinuationListener() {
@Override
public void onComplete(Continuation continuation) {
PushAdapter adapter = (PushAdapter) continuation.getAttribute("adapter");
if (null != adapter) {
continuation.setAttribute("adapter", null);
}
}
@Override
public void onTimeout(Continuation continuation) {
onComplete(continuation);
}
});
resp.flushBuffer();
}
if (continuation.isExpired()) {
return;
}
Writer writer = getWriter(resp);
PushAdapter adapter = (PushAdapter) continuation.getAttribute("adapter");
Message message;
while (true) {
message = adapter.getPushAgent().pull();
if (null == message)
break;
try {
writer.write(message.getContent());
writer.flush();
writer.write("\r\n");
writer.flush();
resp.flushBuffer();
} catch (Exception e) {
throw e;
}
}
//若没有该客户端的消息,则请求被挂起
            continuation.suspend();
}
}
private Writer getWriter(HttpServletResponse response) throws IOException {
OutputStream os = response.getOutputStream();
return new OutputStreamWriter(os, "UTF-8");
}

}
　　向客户端推送消息的servlet:



package com.jiaoyiping.websample.asyncServlet.jetty;
import javax.servlet.ServletException;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.Map;
/*
* Created with Intellij IDEA
* USER: 焦一平
* Mail: jiaoyiping@gmail.com
* Date: 2016/10/25
* Time: 23:46
* To change this template use File | Settings | Editor | File and Code Templates
*/
@WebServlet(urlPatterns = "/send")
public class MesssageSendServlet extends HttpServlet {
@Override
protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
//不要在自己实现的servlet中调用 super.doGet(0)或者是super.doPost()
//因为在tomcat它们的默认实现是报405(HTTP1.1)或者400(其他版本的HTTP)
//        super.doPost(req, resp);

String target = req.getParameter("target");
String messageStr = req.getParameter("message");
Map<String, PushAgent> agentMap = (Map<String, PushAgent>) req.getServletContext().getAttribute("agentmap");
if (agentMap.keySet().contains(target)) {
Message message = new Message();
message.setTarget(target);
message.setContent(messageStr);
if (agentMap.get(target).isInited()) {
agentMap.get(target).onEvent(message);
}
agentMap.get(target).send(message);
PrintWriter out = resp.getWriter();
out.print("发送成功");
out.flush();
out.close();
}

}
@Override
protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
this.doPost(req, resp);
}
}
　　推送代理:就是可以拿到客户端相关信息,并且维护客户端消息队列的类,在这个推送代理中,我们可以加入一个监听器,当有数据需要推送的时候,激活请求



package com.jiaoyiping.websample.asyncServlet.jetty;
/*
* Created with Intellij IDEA
* USER: 焦一平
* Mail: jiaoyiping@gmail.com
* Date: 2016/10/25
* Time: 22:56
* To change this template use File | Settings | Editor | File and Code Templates
*/
public interface PushAgent {
Terminal getTerminal();
String getAddress();
String getToken();
Message send(Message message);
Message pull();
void addListener(MessageListener messageListener);
void onEvent(Message message);
boolean isInited();
}
　　默认实现(每个需要接受推送的用户对应一个PushAgent,和用户端保持长连接的线程从queue里读取mesage对象,向某个用户推送的时候将message对象放到该用户对应的PushAgent的queue里,这里是一个生产者-消费者模式):



package com.jiaoyiping.websample.asyncServlet.jetty;
/*
* Created with Intellij IDEA
* USER: 焦一平
* Mail: jiaoyiping@gmail.com
* Date: 2016/10/25
* Time: 23:17
* To change this template use File | Settings | Editor | File and Code Templates
*/
import java.util.PriorityQueue;
import java.util.Queue;
public class DefaultPushAgent implements PushAgent {
private Terminal terminal;
//客户端通过长连接连接到服务器时,服务器不断地从该队列poll(),若果拿到新的消息,则返回给客户端
private Queue<Message> messages = new PriorityQueue<>();
private MessageListener messageListener;
@Override
public Terminal getTerminal() {
return this.terminal;
}
@Override
public String getAddress() {
return null;
}
@Override
public String getToken() {
return null;
}
@Override
public Message send(Message message) {
synchronized (message) {
messages.add(message);
}
return message;
}
@Override
public Message pull() {
synchronized (messages) {
return messages.poll();
}
}
@Override
public void addListener(MessageListener messageListener) {
this.messageListener = messageListener;
}
@Override
public void onEvent(Message message) {
this.messageListener.onMessage(message);
}
@Override
public boolean isInited() {
return this.messageListener != null;
}

public DefaultPushAgent(Terminal terminal) {
this.terminal = terminal;
}
}

PushAdapter的实现(用户将Continuation和PushAgent关联起来):


package com.jiaoyiping.websample.asyncServlet.jetty;
/*
* Created with Intellij IDEA
* USER: 焦一平
* Mail: jiaoyiping@gmail.com
* Date: 2016/10/25
* Time: 23:37
* To change this template use File | Settings | Editor | File and Code Templates
*/
import org.eclipse.jetty.continuation.Continuation;
public class PushAdapter {
private Continuation continuation;
private PushAgent pushAgent;
public PushAdapter(Continuation continuation, PushAgent pushAgent) {
this.continuation = continuation;
this.pushAgent = pushAgent;
this.pushAgent.addListener(message -> {
if (PushAdapter.this.continuation.isSuspended()) {
PushAdapter.this.continuation.resume();
}
});
}
public Continuation getContinuation() {
return continuation;
}
public void setContinuation(Continuation continuation) {
this.continuation = continuation;
}
public PushAgent getPushAgent() {
return pushAgent;
}

public void setPushAgent(PushAgent pushAgent) {
this.pushAgent = pushAgent;
}
}
　　MessageListener的实现(监听需要推送消息的事件,这里为了做演示,并没有实现一个完整的观察者模式,只是在需要推送消息的时候,手工调用 onMessage()):



package com.jiaoyiping.websample.asyncServlet.jetty;
/*
* Created with Intellij IDEA
* USER: 焦一平
* Mail: jiaoyiping@gmail.com
* Date: 2016/10/26
* Time: 2:03
* To change this template use File | Settings | Editor | File and Code Templates
*/
public interface MessageListener {
void onMessage(Message message);
}   
　　测试数据:使用一个listener在应用初始化的时候,初始化一些数据做为测试数据



package com.jiaoyiping.websample.asyncServlet.jetty;
import javax.servlet.ServletContextEvent;
import javax.servlet.ServletContextListener;
import javax.servlet.annotation.WebListener;
import java.util.HashMap;
import java.util.Map;
/*
* Created with Intellij IDEA
* USER: 焦一平
* Mail: jiaoyiping@gmail.com
* Date: 2016/10/25
* Time: 23:55
* To change this template use File | Settings | Editor | File and Code Templates
*/
@WebListener
public class PushListener implements ServletContextListener {
@Override
public void contextInitialized(ServletContextEvent sce) {
Map<String, PushAgent> agentMap = new HashMap<>();
agentMap.put("zhangsan", new DefaultPushAgent(new Terminal() {{
setAddress("zhangsan");
setToken("zhangsan_token");
}}));
agentMap.put("lisi", new DefaultPushAgent(new Terminal() {{
setAddress("lisi");
setToken("lisi_token");
}}));
sce.getServletContext().setAttribute("agentmap",agentMap);
}
@Override
public void contextDestroyed(ServletContextEvent sce) {
}
}
　　最终的效果是这样的,我截了一个git图: