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

     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，放到你定义的一个全局容器中，比如这样

protected ArrayList connections = 
        new ArrayList();   
synchronized(connections) { 
                connections.add(response); 
            } 

 
 

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

EventType.READ:这表明你可以读用户的消息，并且消息是有效地，并且告诉你这时候读该用户消息是不会有阻塞的风险，当你不幸遇到了读取错误，将会抛出一个异常，这时候会转到下面的EventType.ERROR ，然后这个连接将被关闭，但是你也可以catch这个异常。在window中，一个客户端断开可能是由于一个read事件，读取流可能导致返回-1,IOException，或者 EOFException.确保你处理了以上三种情况，否则你就会跳到EventType.ERROR事件中。

EventType.END: 一次用户请求结束，也就是一次request，服务器已经发回了response,rquest,response没有被回收，但是假如客户端浏览器断开这个comet.这样我们前面那个response的容器就需要清除该response,

 

synchronized(connections) { 
                connections.remove(response); 
            } 

 

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

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

public class ChatServlet 
    extends HttpServlet implements CometProcessor { 

    protected ArrayList connections = 
        new ArrayList(); 
    protected MessageSender messageSender = null; 
    
    public void init() throws ServletException { 
//聊天服务器servlet启动的时候自动启动一个线程来接收用户的聊天消息，并广播出去 
        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; 
    } 

    /** 
     * Process the given Comet event. 
     * 
     * @param event The Comet event that will be processed 
     * @throws IOException 
     * @throws ServletException 
     */ 
    public void event(CometEvent event) 
        throws IOException, ServletException { 
        HttpServletRequest request = event.getHttpServletRequest(); 
        HttpServletResponse response = event.getHttpServletResponse(); 
        if (event.getEventType() == CometEvent.EventType.BEGIN) { 
//加入刚收到这个用户的请求，触发CometEvent.EventType.BEGIN事件，先打印出一些消息头，并且把这个用户的//response保存在缓存容器中，以备广播用 
            log("Begin for session: " + request.getSession(true).getId()); 
            PrintWriter writer = response.getWriter(); 
            writer.println(""-//w3c//dtd html 4.0 transitional//en\">"); 
            writer.println("JSP Chat"); 
            writer.flush(); 
            synchronized(connections) { 
                connections.add(response); 
            } 
        } 
假如IO错误了，当然释放这个连接，清空这个缓存的response句柄，该用户web im应该提示服务器错误，或者超时 
else if (event.getEventType() == CometEvent.EventType.ERROR) { 
            log("Error for session: " + request.getSession(true).getId()); 
            synchronized(connections) { 
                connections.remove(response); 
            } 
            event.close(); 
        } 
//结束时候也是释放连接，清空这个缓存的response句柄，值得注意的是，这时候用户已经断开聊天服务器 

else if (event.getEventType() == CometEvent.EventType.END) {
            log("End for session: " + request.getSession(true).getId());
            synchronized(connections) {
                connections.remove(response);
            }
            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//由于已经在线程中已经读消息和广播消息了，这里这个事件主要用来log输出用户发来的聊天内容
                if (n > 0) {
                    log("Read " + n + " bytes: " + new String(buf, 0, n) 
                            + " for session: " + request.getSession(true).getId());
                } else if (n < 0) {
                    error(event, request, response);
                    return;
                }
            } while (is.available() > 0);
        }
    }

 

 

//该线程用来接收用户的消息，和向所有用户广播消息,应该不是很难

    public class MessageSender implements Runnable {

        protected boolean running = true;
        protected ArrayList<String> messages = new ArrayList<String>();
        
        public MessageSender() {
        }
        
        public void stop() {
            running = false;
        }

        /**
         * Add message for sending.
         */
        public void send(String user, String message) {
            synchronized (messages) {
                messages.add("[" + user + "]: " + message);
                messages.notify();
            }
        }

        public void run() {

            while (running) {

                if (messages.size() == 0) {
                    try {
                        synchronized (messages) {
                            messages.wait();
                        }
                    } catch (InterruptedException e) {
                        // Ignore
                    }
                }

                synchronized (connections) {
                    String[] pendingMessages = null;
                    synchronized (messages) {
                        pendingMessages = messages.toArray(new String[0]);
                        messages.clear();
                    }
                    // 传说中的广播
                    for (int i = 0; i < connections.size(); i++) {
                        try {
                            PrintWriter writer = connections.get(i).getWriter();
                            for (int j = 0; j < pendingMessages.length; j++) {
                                writer.println(pendingMessages[j] + "<br>");
                            }
                            writer.flush();
                        } catch (IOException e) {
                            log("IOExeption sending message", e);
                        }
                    }
                }

            }

        }

    }

}

 

 

评论

16 楼 rmn190 2011-01-21  

有没有人用过Tomcat6中支持的Asynchronous writes？

15 楼 shuchang5920 2010-07-12  

这东西对我来说实在是太陌生了，我需要在学习一段时间，感谢楼主提供的示例！

14 楼 wy19280469 2010-07-06  

你应该了解
tomcat6 的nio 就是异步的

13 楼 spell 2010-07-06  

这个方式很好吗？我还是喜欢用socket + flash的解决方案，flash是支持socket的，又可以放在网页里面，所以这个很好使用。有谁知道说说comet的原理啊？

12 楼 javagui 2010-07-05  

jetty N年前（那时候servlet 3.0连草案还没有）就实现了。

11 楼 pxlfxl2 2010-07-05  

IcyFenix 写道

说实话，我真的很怀疑现在的Tomcat7是不是能达到Beta的程度，很多很基本的功能点，在glassfish v3上运行的很好，在tomcat7上就会出莫名其妙的问题。

所以要体验servlet3的话，现阶段建议还是拿glassfish来玩。

非常同意楼上的观点，现在Tomcat 7.0号称是Beta版，其实很不稳定，Bug很多，连jsp都不能很好的兼容，解析时总是报莫名其妙的xml解析异常，上周整了一天，差点没把我整死，这些在glassfish3下不不会出现的。glassfish确实非常强大，就怪sun的销售做的太烂，这么好的产品居然赚不到钱，定制了java ee标准，结果钱都让别人赚了，市场也丢了，现在公司都没了，唉。。。。

10 楼 kakaluyi 2010-07-04  

IcyFenix 写道

说实话，我真的很怀疑现在的Tomcat7是不是能达到Beta的程度，很多很基本的功能点，在glassfish v3上运行的很好，在tomcat7上就会出莫名其妙的问题。

所以要体验servlet3的话，现阶段建议还是拿glassfish来玩。

恩 最近在公司没事做准备改进下原先轮询的web im，多谢推荐，下周试试glassfish来实现

9 楼 IcyFenix 2010-07-04  

说实话，我真的很怀疑现在的Tomcat7是不是能达到Beta的程度，很多很基本的功能点，在glassfish v3上运行的很好，在tomcat7上就会出莫名其妙的问题。

所以要体验servlet3的话，现阶段建议还是拿glassfish来玩。

8 楼 kakaluyi 2010-07-04  

恩失误了，上次看到servlet3.0支持异步io,tomcat7又说完全支持servlet3.0，所以以为原生支持comet是tomcat7的特性

7 楼 kakaluyi 2010-07-04  

IcyFenix 写道

1.CometProcessor在Tomcat6的时候就有了，那时候叫高级异步IO特性，这个和JSR315中的异步Servlet本质上就是一回事，但是使用Tomcat7.0的进步是让我们可以直接使用Servlet3.0的API，写出来的程序可以在任何一个支持Servlet3.0的Web容器上运行，而不用去实现org.apache.catalina.CometProcessor这种接口绑死在Tomcat上。

2.LZ的文章并非原创，起码代码实例部分应该来自这里吧：http://www.ibm.com/developerworks/cn/web/wa-cometjava/

恩我的代码部分确实不是原创的，但是加上自己的理解，这篇imb文章引用的代码也不是原创，都是引用apche7.0 comet的demo代码（原文英文），我这篇文章有很多自己分析的东西，所以理解也不一定准确

6 楼 hantsy 2010-07-04  

tomcat 7.0 实现 servlet3.0 本来就运行异步请求和 comet 无关。。。

5 楼 IcyFenix 2010-07-04  

1.CometProcessor在Tomcat6的时候就有了，那时候叫高级异步IO特性，这个和JSR315中的异步Servlet本质上就是一回事，但是使用Tomcat7.0的进步是让我们可以直接使用Servlet3.0的API，写出来的程序可以在任何一个支持Servlet3.0的Web容器上运行，而不用去实现org.apache.catalina.CometProcessor这种接口绑死在Tomcat上。

2.LZ的文章并非原创，起码代码实例部分应该来自这里吧：http://www.ibm.com/developerworks/cn/web/wa-cometjava/

4 楼 liubobo 2010-07-04  

楼上的说的对，tomcat7.0 beta版对comet的支持没有改变多大，跟tomcat6.0是一样的，只是CometProcessor的包被换了一下而已。我已经用tomcat6.0的comet支持实现了一对一、一对多的及时通讯。希望对comet感兴趣的朋友共同交流一下。等tomcat7.0正式版出来后，comet的支持会比较方便一些。

3 楼 pxlfxl2 2010-07-04  

这个不是7.0的新特新，6.0就有了，7.0主要还是对java EE6部分新标准的实现，主要是Servlet 3.0、JSP 2.2及EL 2.2

2 楼 sky3380 2010-07-04  

再排下版吧，看着有点乱

1 楼 southgate 2010-07-03  

6.0的时候tomcat已经有自带的CometProcessor了把