|
|
在Tomcat 6中，增加了对Comet的支持。原理很简单，服务器端不关闭respose，并源源不断的给浏览器发送数据流
Java 有一个丰富的本地线程模型。因此实现典型的每个连接一个线程的模型便非常简单。该模型对于 Comet 不大适用，但是，Java 对此同样有解决的办法。为了有效地处理 Comet，需要非阻塞 IO，Java 通过它的 NIO 库提供非阻塞 IO.两种最流行的开源服务器 Apache Tomcat 和 Jetty 都利用 NIO 增加非阻塞 IO，从而支持 Comet.
使用tomcat提供的comet api之前，需要增加对NIO的支持，配置nio可是相当方便，要做的仅仅是在server.xml里边修改connector：
<connector protocol="org.apache.coyote.http11.Http11NioProtocol" port="8080" redirectport="8443" connectiontimeout="20000">
Tomcat 6提供了CometProcessor和CometEvent接口，实现CometProcessor接口后不用在servlet中写doGet,doPoset方法，所有事件在BEGIN,READ,END,ERROR中实现。
CometProcessor 接口要求实现 event 方法。这是用于 Comet 交互的一个生命周期方法。Tomcat 将使用不同的 CometEvent 实例调用。通过检查 CometEvent 的 eventType，可以判断正处在生命周期的哪个阶段。当请求第一次传入时，即发生 BEGIN 事件。READ 事件表明数据正在被发送，只有当请求为 POST 时才需要该事件。遇到 END 或 ERROR 事件时，请求终止。
一般采用的方法是，客户端先在实现CometProcessor的servlet中进行注册，保存HttpServletResponse，应用将通知事件放入处理的队列，实现CometProcessor的servlet中建立线程组，对需要处理的队列里的时间进行处理，并将结果通过HttpServletResponse异步发送给需要发送给客户端
事例代码
public class ChatServlet extends HttpServlet implements CometProcessor {
private static final long serialVersionUID = -3667180332947986301L;
// <用户,长连接>
protected static Map<String, HttpServletResponse> connections = new HashMap<String, HttpServletResponse>();
// 消息推送线程
protected static MessageSender messageSender = null;
public void init() throws ServletException {
System.out.println("-----------------");
// 启动消息推送线程
messageSender = new MessageSender();
Thread messageSenderThread = new Thread(messageSender, "MessageSender["
+ getServletContext().getContextPath() + "]");
messageSenderThread.setDaemon(true);
messageSenderThread.start();
}
public void destroy() {
connections.clear();
messageSender.stop();
messageSender = null;
}
public void event(CometEvent event) throws IOException, ServletException {
HttpServletRequest request = event.getHttpServletRequest();
HttpServletResponse response = event.getHttpServletResponse();
// 昵称
String name = request.getParameter("name");
if (name == null) {
return;
}
if (event.getEventType() == CometEvent.EventType.BEGIN) {
// Http连接空闲超时
event.setTimeout(Integer.MAX_VALUE);
log("Begin for session: " + request.getSession(true).getId());
// 创建Comet Iframe
PrintWriter writer = response.getWriter();
writer
.println("<!doctype html public \"-//w3c//dtd html 4.0 transitional//en\">");
writer
.println("<html><head><script type=\"text/javascript\">var comet = window.parent.comet;</script></head><body>");
writer.println("<script type=\"text/javascript\">");
writer.println("var comet = window.parent.comet;");
writer.println("</script>");
writer.flush();
// for chrome
if (request.getHeader("User-Agent").contains("KHTML")) {
for (int i = 0; i < 100; i++) {
writer.print("<input type=hidden name=none value=none>");
}
writer.flush();
}
// 欢迎信息
writer.print("<script type=\"text/javascript\">");
writer.println("comet.newMessage('Hello " + name + ", Welcome!');");
writer.print("</script>");
writer.flush();
// 通知其他用户有新用户登陆
if (!connections.containsKey(name)) {
messageSender.login(name);
}
// 推送已经登陆的用户信息
for (String user : connections.keySet()) {
if (!user.equals(name)) {
writer.print("<script type=\"text/javascript\">");
writer.println("comet.newUser('" + user + "');");
writer.print("</script>");
}
}
writer.flush();
synchronized (connections) {
connections.put(name, response);
}
} else if (event.getEventType() == CometEvent.EventType.ERROR) {
log("Error for session: " + request.getSession(true).getId());
synchronized (connections) {
connections.remove(name);
}
event.close();
} else if (event.getEventType() == CometEvent.EventType.END) {
log("End for session: " + request.getSession(true).getId());
messageSender.logout(name);
synchronized (connections) {
connections.remove(name);
}
PrintWriter writer = response.getWriter();
writer.println("</body></html>");
event.close();
} else if (event.getEventType() == CometEvent.EventType.READ) {
InputStream is = request.getInputStream();
byte[] buf = new byte[512];
do {
int n = is.read(buf); // can throw an IOException
if (n > 0) {
log("Read " + n + " bytes: " + new String(buf, 0, n)
+ " for session: "
+ request.getSession(true).getId());
} else if (n < 0) {
return;
}
} while (is.available() > 0);
}
}
// 发送消息给所有人
public static void send(String message) {
messageSender.send("*", message);
}
// 向某个连接发送消息
public static void send(String name, String message) {
messageSender.send(name, message);
}
public class MessageSender implements Runnable {
protected boolean running = true;
protected Map<String, String> messages = new HashMap<String, String>();
public MessageSender() {
}
public void stop() {
running = false;
}
// 新用户登陆
public void login(String name) {
synchronized (messages) {
messages.put("Login", name);
messages.notify();
}
}
// 用户下线
public void logout(String name) {
synchronized (messages) {
messages.put("Logout", name);
messages.notify();
}
}
// 发送消息
public void send(String user, String message) {
synchronized (messages) {
messages.put(user, message);
messages.notify();
}
}
public void run() {
while (running) {
if (messages.size() == 0) {
try {
synchronized (messages) {
messages.wait();
}
} catch (InterruptedException e) {
// Ignore
}
}
synchronized (connections) {
synchronized (messages) {
// 推送消息队列中的消息
for (Entry<String, String> message : messages
.entrySet()) {
if (message.getKey().equals("Login")) {// 新用户登陆
log(message.getValue() + " Login");
for (HttpServletResponse response : connections
.values()) {
try {
PrintWriter writer = response
.getWriter();
writer
.print("<script type=\"text/javascript\">");
writer
.println("comet.newMessage('Welcome "
+ message.getValue()
+ " !');");
writer.println("comet.newUser('"
+ message.getValue() + "');");
writer.print("</script>");
writer.flush();
} catch (IOException e) {
log("IOExeption execute command", e);
}
}
} else if ("Logout".equals(message.getKey())) {// 用户退出
log(message.getValue() + " Logout");
for (HttpServletResponse response : connections
.values()) {
try {
PrintWriter writer = response
.getWriter();
writer
.print("<script type=\"text/javascript\">");
writer.println("comet.newMessage('88, "
+ message.getValue() + "');");
writer.println("comet.deleteUser('"
+ message.getValue() + "');");
writer.print("</script>");
writer.flush();
} catch (IOException e) {
log("IOExeption execute command", e);
}
}
} else if ("*".equals(message.getKey())) {// 群发消息
log("Send message: " + message.getValue()
+ " to everyone.");
for (HttpServletResponse response : connections
.values()) {
try {
PrintWriter writer = response
.getWriter();
writer
.print("<script type=\"text/javascript\">");
writer.println("comet.newMessage('"
+ message.getValue() + "');");
writer.print("</script>");
writer.flush();
} catch (IOException e) {
log("IOExeption execute command", e);
}
}
} else {// 向某人发信息
try {
HttpServletResponse response = connections
.get(message.getKey());
PrintWriter writer = response.getWriter();
writer
.print("<script type=\"text/javascript\">");
writer.println("comet.privateMessage('"
+ message.getValue() + "');");
writer.print("</script>");
writer.flush();
} catch (IOException e) {
log("IOExeption sending message", e);
}
}
// 从消息队列中删除消息
messages.remove(message.getKey());
}
}
}
}
}
}
}
connections 就是保存HttpServletResponse的list，MessageSender 就是处理推送的线程，MessageSender 中的messages ，存储的就是呆发送的消息。需要推送给客户端时，从connections 中取出需要的HttpServletResponse，将信息writer进去进可以了 |
|
|
QQ群⑧：
窥视卡
雷达卡
发表于 2017-1-19 07:05:57
提升卡
置顶卡
沉默卡
喧嚣卡
变色卡
千斤顶
显身卡