任何一个想实现专业Comet产品都避免不了去了解Bayeux, 它定义了一个协议，通过一系列 json 的事件来实现 pubsub 模型进而实现Comet各种业务。它也致力让自己成为Comet的标准。小型的Comet应用通常会在Streaming里面返回一些自定义的 JavaScript来实现各种业务功能，但是Bayeux实现了一个full stack的体系，包括客户端服务器的交互，Event, Transport协商, Authentication, Security等。

Bayeux一些术语和概念包括

• Pub/Sub,Bayeux的核心就是通过一个pub/sub模型来实现Comet各种业务。
• ClientID, 通过客户端handshake之后服务器分配。
• Channel,订阅的频道，可以理解成聊天的房间。
• Messages: 所有的包都封装成一个message，里面包含各种字段。
• data: message中的数据元素，应用可以根据业务再次定义细分的格式。

Bayeux的优势很明显，通过与Jetty配合可以几十行程序实现一个comet的聊天室demo

缺点

• PubSub模型，点对点支持不佳，除了多人聊天的场合之外，其实大部分comet应用更适合点对点的模式。如果用PubSub套上去也可以实现，问题是对服务器的压力更高，而且通常comet应用的访问量都比较大。
• 可靠性，一条信息publish了没法知道客户是否能收到了，也没法做进一步处理比如说保存或重发等业务逻辑。
• 顺序，publisher顺序和接收显示的顺序
• 多Tab浏览器打开同一个Comet应用，或者在IE里面Ctrl-N再次打开当前页面新窗口。

更详细的比较可参看 Battle of the Bayeux 系列

尽管Bayeux喜忧参半，但是对于希望实现一个专业Comet应用的开发者来说，Bayeux的诱惑是巨大的，它有开发者需要的完备的接口和成熟的体系架构。如果绕开它，不管是服务器还是客户端都需要一个漫长的摸索和自己定义规范的过程。如果看了这篇自己用C实现的comet http server,基于libevent以为就搞定Comet那可能就低估难度了。

Bayeux的例子可以通过下载Jetty6看到。

前不久自己用C实现的comet http server,基于libevent，用c实现了一个comet server原型，由于libevent封装了一个http协议实现，代码比较简洁，但不简单。最近又了解了一下Java的实现方案。

• Java中目前有2种选择，一种是基于Apache MINA框架的，稍复杂但适合用在大场合；一种是基于Java Web容器自己提供的支持，如Jetty和Tomcat，适合comet只是应用中一小部分场合。
• Mina只是个网络层(相当socket层)的框架，如果用mina需要自己实现HTTP协议。好消息是这个工作已经有开源做了，一个叫Asyncweb项目在mina的基础上实现了HTTP协议层的封装。而且asyncweb正在被mina合并，将加入到mina的发行版中。

用asyncweb编程很简单，只需实现一个接口，实际上只需要实现 handleRequest, 在 request 返回结果。

public interface HttpService
{
    void handleRequest( HttpServiceContext context ) throws Exception;

    void start();

    void stop();
}

• 经过测试，asyncweb适合做异步long pooling,但不适合做Streaming XHR, 参看一下例子

handleRequest 例子，不管是同步还是异步，都必须一步输出，不支持一次写一部分

/** 同步处理的例子 */
public void handleRequest( HttpServiceContext context ) throws Exception {
    MutableHttpResponse response = new DefaultHttpResponse();

    StringWriter buf = new StringWriter();
    PrintWriter writer = new PrintWriter(buf);
    writer.println("test");
    writer.flush();

    IoBuffer bb = IoBuffer.allocate(1024);
    bb.setAutoExpand(true);
    bb.putString(buf.toString(), Charset.forName("UTF-8").newEncoder());
    bb.flip();
    response.setContent(bb);

    response.setHeader("Pragma", "no-cache");
    response.setHeader("Cache-Control", "no-cache");
    response.setStatus(HttpResponseStatus.OK);

    context.commitResponse(response);
}

/** 异步处理的例子，可以不马上返回，但是必须一次返回 */
public void handleRequest( HttpServiceContext context ) throws Exception {
    context.addClientListener( new HttpClientListener() {
        public void clientDisconnected( HttpServiceContext ctx ) {
        }

        public void clientIdle( HttpServiceContext ctx, long idleTime, int idleCount ) {
            // do something...
            // context.commitResponse(...)
        }
    });
}

所以Asyncweb如果需要Streaming XHR功能还需要修改source code，所以目前成熟的做法还是用Jetty or Tomcat，另外在网上见到说GlassFish中的Grizzly使用修改了asyncweb实现了comet功能，或许那个1年多前的修改版本更值得借鉴。

Jetty 的Comet实现从编程的角度是最简单的，suspend的时候本次执行就结束了。

public class ForeverFrameJettyServlet extends HttpServlet {
   public void service(HttpServletRequest req, HttpServletResponse resp) throws IOException {
      // handle the time request
      if (req.getRequestURI().endsWith("/time")) {
         // get the jetty continuation
         Continuation cc = ContinuationSupport.getContinuation(req, null);

         // set the header
         resp.setContentType("text/html");

         // write time periodically
         while (true) {
            cc.suspend(1000); // suspend the response
            resp.getWriter().println("test");
            resp.getWriter().flush();
         }
      }

      // ...
   }
}

Tomcat6 也支持Comet,需要实现一个CometProcessor接口，稍繁琐。例子可参考下面Resource1

Resource:
1. Asynchronous HTTP and Comet architectures
An introduction to asynchronous, non-blocking HTTP programming
http://www.javaworld.com/javaworld/jw-03-2008/jw-03-asynchhttp.html

2. AsyncWeb Performance Test 与Apache2性能比较

comet http server是一个非常简单的服务器，它只是一个通道管理器，把application server需要发送的信息送到browser。application server通常在别的服务器，可能采用任何语言任何操作系统开发。他们之间必须选择一种方式通讯。如果是同一种语言通讯就比较简单，但是异构系统通常要用web service等方法来实现，比较低效。memcache协议是一种更高效的方式，同时具有各种语言的client library。

图示如下，是一个share nothing的架构，comet server之间不需要通讯
(这个图是 orbited.org 作者画的，他实现了一个python的comet方案)

至于memcache协议层的server实现就可以找现成的，比如用memcached的code修改。前不久刚好下载了sina开发的memcachedb，封装得也很好，就一个memcachedb.c文件。只要把自己用C实现的comet http server,基于libevent的源代码合并进去即可。当然它里面的bdb部分也不需要了，可以全部删掉。因为只需要他的协议处理部分。

修改部分的Source code

/* add function for send to comet */
void notify_comet(char *key, char *str) {
    int nudge = atoi(key);
    if (requests[nudge] != NULL) {
        evbuffer_add_printf(bufs[nudge], "Oops, from memcache: %s!\r\n", str);
        evhttp_send_reply_chunk(requests[nudge], bufs[nudge]);
    }
}

/* 
 * we get here after reading the value in set/add/replace commands. The command
 * has been stored in c->item_comm, and the item is ready in c->item.
 * 这个函数完全修改了
 */
void complete_nread(conn * c) {
    item *it = &(c->item);

    int comm = c->item_comm;

    int ret;

    stats.set_cmds++;

    while (1) {
        if (strncmp(ITEM_data(it) + it->nbytes - 2, "\r\n", 2) != 0) {
            out_string(c, "CLIENT_ERROR bad data chunk");
            break;
        }
        out_string(c, "STORED");

        // send to comet server
        notify_comet(ITEM_key(it), ITEM_data(it));
        break;
    }

    return;
}

• 基于libevent 1.3e, 利用了libevent里面现成的http库
• 编译方法，先安装libevent,然后
      cc -o httpd httpd.c -L/usr/local/lib/ -Wall -O2 -I/usr/local/include -ggdb -levent
• 可以作为一个商业运行环境的基础，真正的业务系统里面，comet server也很简单，比这个程序复杂不了什么，它只处理comet，只是一个通道，复杂的东西放到其他地方。
• 打算作为下图中的一个模块

Source code:

#include <sys/types.h>
#include <sys/time.h>
#include <stdlib.h>
#include <err.h>

#include <sys/queue.h>
#include <evhttp.h>
#define MAX_CONN 10240

struct evhttp_request *requests[MAX_CONN];
struct evbuffer *bufs[MAX_CONN];
int pos[MAX_CONN], num;

void on_close(struct evhttp_request *req, void *arg) {
 int* p = (int*) arg;
 struct evhttp_request *p2 = requests[*p];
 struct evbuffer *p3 = bufs[*p];
 printf("%s closed.\n", p2->remote_host);
 requests[*p] = NULL;
 if (p3 != NULL) evbuffer_free(p3);
}

/* for comet recv */
void comet_handler(struct evhttp_request *req, void *arg) {
 int i, j;
 requests[num] = req;
 struct evbuffer *buf; 
  
 req->minor = 0; 
 evhttp_add_header(req->output_headers, "Content-Type", "text/plain");

    evhttp_send_reply_start(req, HTTP_OK, "OK");
 
 /* Set a callback for connection close. */
 evhttp_connection_set_closecb(req->evcon, on_close, &pos[num]);

 buf = evbuffer_new();
 bufs[num] = buf;

 for (j = 0; j < 10; j++)
  evbuffer_add_printf(buf, "Hi %s, this is a comet server based on libevent, visit http://hi.baidu.com/jabber for more information!\r\n", req->remote_host);
    evhttp_send_reply_chunk(req, buf);

    for (i = 0; i < num; i++) {
     if (requests[i] == NULL) continue;
     evbuffer_add_printf(bufs[i], "Oops, %s requests!\r\n", req->remote_host);
     evhttp_send_reply_chunk(requests[i], bufs[i]);
     // evhttp_send_reply_end(requests[i]);
    }
    
    if (num++ == MAX_CONN) err(1, "exceed max conn, TODO.");
}

/* for comet send, GET /id */
void notify_handler(struct evhttp_request *req, void *arg) {
    struct evbuffer *buf;
    int nudge;
    
    buf = evbuffer_new();
    if (buf == NULL) err(1, "failed to create response buffer");
    const char *uri = evhttp_request_uri(req);
    evbuffer_add_printf(buf, "Requested: %s, send a nudge to %s\n", uri, uri + 1);

 nudge = atoi(uri + 1);
 if (requests[nudge] != NULL) {
  evbuffer_add_printf(bufs[nudge], "Oops, received a nudge from %s!\r\n", req->remote_host);
  evhttp_send_reply_chunk(requests[nudge], bufs[nudge]);
 }
    
    evhttp_send_reply(req, HTTP_OK, "OK", buf); // TODO: free buf
}

int main(int argc, char **argv) {
        struct evhttp *httpd;
        int i;
        for (i = 0; i < 1024; i++) {
         requests[i] == NULL;
         bufs[i] = NULL;
         pos[i] = i;
        }
        num = 0;

        event_init();
        httpd = evhttp_start("0.0.0.0", 8080);

        /* Set a comet callback for requests to "/". */
        evhttp_set_cb(httpd, "/", comet_handler, NULL);

        /* Set a send callback for all other requests. */ 
        evhttp_set_gencb(httpd, notify_handler, NULL);

        event_dispatch();

        /* Not reached in this code as it is now. */
        evhttp_free(httpd);
        return 0;
}

参考：
http://blog.gslin.org/archives/2005/11/24/220/ libevent编程介绍，里面的留言较搞笑
http://unx.ca/log/tag/libevent/ 比较旧了，基于libevent 1.2 的例子
http://www.slideshare.net/simon/time-for-comet/ 什么是 comet，这里有原理说明

看到了另外一个基于Web的IM实现思考，不过不太认同里面的观点。我的一些看法如下。

• IM 除语音外，文字信息对一两秒左右的延迟并不敏感，语音通常对实时性要求高，但语音通常是VOIP研究的范围，只是IM一个附加功能。
• 主流的交互式网络应用都是基于服务器的，如网游。
• IM除了文字信息通讯之外，还有两个主要的功能，presence和roster管理，不管客户端怎样p2p,这两个信息通常都是服务器来管理。所以使用web还是客户端并无区别。相反使用客户端这两个信息通常只有厂商自身的软件才能提供，如果使用web可以方便嵌入任何页面和应用。
• 以客户端为代表的IM如QQ已经发展到一个极限，他们通常自成一个体系，无法利用web互联互通的优势。他们理解了互联网，但是没有理解web(and web 2.0)。后继的IM厂商仍然没有脱离QQ的思维。同时网络应用提供商无法把IM服务像webmail，blog，bbs一样提供给自己的客户。
• IM应当象facebook api一样成为一种服务。

示意图说明：

• 蓝色的箭头是发送，红色的是接收
• 利用connection manager来实现multiplex
• 每个IM用户用一个WebIMSession来保存会话信息
• 到openfire之间连接很少，通常5个
• 收到消息之后通过WebIMSession定位到用户，触发ajax listener到界面弹出信息
• 瓶颈在web ajax connections / web server