利用Tomcat7.0新特性，用BS实现CS的聊天服务器

teddylw1611616

Tomcat7.0.0已经出来了，关注它已经支持servlet3.0规范，servlet3.0规范有一个很值得期待的特性就是，支持异步IO通信，何为异步响应，就是保持长连接，让servlet实现原先的TCP Server才能做到的事，就像我以前写的一个WEB IM。没有用到comet,使用Ajax轮询聊天，反应慢不说，很多时候轮询的资源是被浪费掉的,杯具啊

    现在好了Tomcat7.0.0已经原生支持comet和异步IO，但是需要APR 或者NIO HTTP连接器，应该在新的servlet-api,会提供tomcat7.0-guide的原文如下：

Usage of these features requires using the APR or NIO HTTP connectors. The classic java.io HTTP connector and the AJP connectors do not support them，
实际是使用,需要增加对NIO的支持，要做的仅仅是在server.xml里边修改connector：

<connector protocol="org.apache.coyote.http11.Http11NioProtocol" port="8080" redirectport="8443" connectiontimeout="20000">


    要用好这个comet需要了解comet的几个事件


EventType.BEGIN:开始连接，比如用户的一个聊天消息刚发送到了你的comet聊天服务器，建立连接后，但是你还没有开始读。（因为被加锁了，有可能正在读其他用户的聊天消息）你可以通过CometEvent 对象获取该用户request,response，当锁被你获得后，使用这些request该怎么做就由你了，比如得到这个request的聊天正文或者头域。最重要的是，你可以取得这个用户的response，放到你定义的一个全局容器中，比如这样

1. protected ArrayList connections =
   
2.         new ArrayList();   
   
3. synchronized(connections) {
   
4.                 connections.add(response);
   
5.             }

复制代码

这端代码很重要，因为只有保存了用户响应的句柄，才是异步comet的关键所在，你可以决定什么时候，什么内容什么顺序把消息分发给哪一个用户.

EventType.READ:这表明你可以读用户的消息，并且消息是有效地，并且告诉你这时候读该用户消息是不会有阻塞的风险，当你不幸遇到了读取错误，将会抛出一个异常，这时候会转到下面的EventType.ERROR ，然后这个连接将被关闭，但是你也可以catch这个异常。在window中，一个客户端断开可能是由于一个read事件，读取流可能导致返回-1,IOException，或者 EOFException.确保你处理了以上三种情况，否则你就会跳到EventType.ERROR事件中。
EventType.END: 一次用户请求结束，也就是一次request，服务器已经发回了response,rquest,response没有被回收，但是假如客户端浏览器断开这个comet.这样我们前面那个response的容器就需要清除该response,

1. synchronized(connections) {
   
2.                 connections.remove(response);
   
3.             }

复制代码

EventType.ERROR:当一次IO异常或者一次不可发生的错误发生，那些在begin方法中初始化的资源会被重置，这个request和response将会被回收(就是某一个发生ioexception客户的资源都要被重置，抓住这个错误可以提示给用户一些有用的信息，比如：你和服务器的连接发生了错误，请重新登录聊天室）


下面这个是模拟TCP CHART Server的Servlet聊天服务器，是基于异步长连接的

1. public class ChatServlet
   
2.     extends HttpServlet implements CometProcessor {
   
3. 
   
4.     protected ArrayList connections =
   
5.         new ArrayList();
   
6.     protected MessageSender messageSender = null;
   
7.    
   
8.     public void init() throws ServletException {
   
9. //聊天服务器servlet启动的时候自动启动一个线程来接收用户的聊天消息，并广播出去
   
10.         messageSender = new MessageSender();
    
11.         Thread messageSenderThread =
    
12.             new Thread(messageSender, &quot;MessageSender[&quot; + getServletContext().getContextPath() + &quot;]&quot;);
    
13.         messageSenderThread.setDaemon(true);
    
14.         messageSenderThread.start();
    
15.     }
    
16. 
    
17.     public void destroy() {
    
18. //清除资源
    
19.         connections.clear();
    
20.         messageSender.stop();
    
21.         messageSender = null;
    
22.     }
    
23. 
    
24.     /**
    
25.      * Process the given Comet event.
    
26.      *
    
27.      * @param event The Comet event that will be processed
    
28.      * @throws IOException
    
29.      * @throws ServletException
    
30.      */
    
31.     public void event(CometEvent event)
    
32.         throws IOException, ServletException {
    
33.         HttpServletRequest request = event.getHttpServletRequest();
    
34.         HttpServletResponse response = event.getHttpServletResponse();
    
35.         if (event.getEventType() == CometEvent.EventType.BEGIN) {
    
36. //加入刚收到这个用户的请求，触发CometEvent.EventType.BEGIN事件，先打印出一些消息头，并且把这个用户的//response保存在缓存容器中，以备广播用
    
37.             log(&quot;Begin for session: &quot; + request.getSession(true).getId());
    
38.             PrintWriter writer = response.getWriter();
    
39.             writer.println(&quot;&quot;-//w3c//dtd html 4.0 transitional//en\&quot;&gt;&quot;);
    
40.             writer.println(&quot;JSP Chat&quot;);
    
41.             writer.flush();
    
42.             synchronized(connections) {
    
43.                 connections.add(response);
    
44.             }
    
45.         }
    
46. 假如IO错误了，当然释放这个连接，清空这个缓存的response句柄，该用户web im应该提示服务器错误，或者超时
    
47. else if (event.getEventType() == CometEvent.EventType.ERROR) {
    
48.             log(&quot;Error for session: &quot; + request.getSession(true).getId());
    
49.             synchronized(connections) {
    
50.                 connections.remove(response);
    
51.             }
    
52.             event.close();
    
53.         }
    
54. //结束时候也是释放连接，清空这个缓存的response句柄，值得注意的是，这时候用户已经断开聊天服务器
    
55. 
    
56. else if (event.getEventType() == CometEvent.EventType.END) {
    
57.             log("End for session: " + request.getSession(true).getId());
    
58.             synchronized(<SPAN style="BACKGROUND-COLOR: yellow">connection</SPAN>s) {
    
59.                 <SPAN style="BACKGROUND-COLOR: yellow">connection</SPAN>s.remove(response);
    
60.             }
    
61.             PrintWriter writer = response.getWriter();
    
62.             writer.println("</body></html>");
    
63.             event.close();
    
64.         } else if (event.getEventType() == CometEvent.EventType.READ) {
    
65.             InputStream is = request.getInputStream();
    
66.             byte[] buf = new byte[512];
    
67.             do {
    
68.                 int n = is.read(buf); //can throw an IOException//由于已经在线程中已经读消息和广播消息了，这里这个事件主要用来log输出用户发来的聊天内容
    
69.                 if (n > 0) {
    
70.                     log("Read " + n + " bytes: " + new String(buf, 0, n)
    
71.                             + " for session: " + request.getSession(true).getId());
    
72.                 } else if (n < 0) {
    
73.                     error(event, request, response);
    
74.                     return;
    
75.                 }
    
76.             } while (is.available() > 0);
    
77.         }
    
78.     }
    
79. 
    

复制代码

1. //该线程用来接收用户的消息，和向所有用户广播消息,应该不是很难
   
2. 
   
3.     public class MessageSender implements Runnable {
   
4. 
   
5.         protected boolean running = true;
   
6.         protected ArrayList<String> messages = new ArrayList<String>();
   
7.         
   
8.         public MessageSender() {
   
9.         }
   
10.         
    
11.         public void stop() {
    
12.             running = false;
    
13.         }
    
14. 
    
15.         /**
    
16.          * Add message for sending.
    
17.          */
    
18.         public void send(String user, String message) {
    
19.             synchronized (messages) {
    
20.                 messages.add("[" + user + "]: " + message);
    
21.                 messages.notify();
    
22.             }
    
23.         }
    
24. 
    
25.         public void run() {
    
26. 
    
27.             while (running) {
    
28. 
    
29.                 if (messages.size() == 0) {
    
30.                     try {
    
31.                         synchronized (messages) {
    
32.                             messages.wait();
    
33.                         }
    
34.                     } catch (InterruptedException e) {
    
35.                         // Ignore
    
36.                     }
    
37.                 }
    
38. 
    
39.                 synchronized (connections) {
    
40.                     String[] pendingMessages = null;
    
41.                     synchronized (messages) {
    
42.                         pendingMessages = messages.toArray(new String[0]);
    
43.                         messages.clear();
    
44.                     }
    
45.                     // 传说中的广播
    
46.                     for (int i = 0; i < connections.size(); i++) {
    
47.                         try {
    
48.                             PrintWriter writer = connections.get(i).getWriter();
    
49.                             for (int j = 0; j < pendingMessages.length; j++) {
    
50.                                 writer.println(pendingMessages[j] + "<br>");
    
51.                             }
    
52.                             writer.flush();
    
53.                         } catch (IOException e) {
    
54.                             log("IOExeption sending message", e);
    
55.                         }
    
56.                     }
    
57.                 }
    
58. 
    
59.             }
    
60. 
    
61.         }
    
62. 
    
63.     }
    
64. 
    
65. }
    
66. 
    

复制代码

原文 http://www.javaeye.com/topic/704973