初识cometd

<!--defaultCSS-->

 

1、项目的基本介绍

官方的解释是一个HTTP事件驱动框架。我没明白这个是什么意思，当前研究它，就是想看看它的消息推送功能。从我的角度看，cometd提供了一套搭建推送消息的框架，既包括server部分，也包括client部分。使用它，我们能够方便、快速的搭建起我们的消息推送系统。

 

2、搭建demo

demo基本介绍：搭建cometd server，开发publish和subscribe消息的客户端，并测试发布和订阅消息

 

1）从cometd-2.6.0-RC1\cometd-java\cometd-java-examples\target\cometd-java-examples-2.6.0-RC1.war包中获取全部依赖的lib库

 

2）创建web工程，用于部署cometd的server服务。

对于我们的demo测试程序，web工程创建好后，不需要写任何代码，只需要配置好web.xml文件，样例如下：

<?xml version="1.0" encoding="UTF-8"?>

<web-app xmlns="http://java.sun.com/xml/ns/javaee"

         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

         xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"

         version="3.0">

    <servlet>

        <servlet-name>cometd</servlet-name>

        <servlet-class>org.cometd.server.CometdServlet</servlet-class>

        <init-param>

            <param-name>timeout</param-name>

            <param-value>60000</param-value>

        </init-param>

        <init-param>

            <param-name>logLevel</param-name>

            <param-value>3</param-value>

        </init-param>

<init-param>

            <param-name>transports</param-name>

            <param-value>org.cometd.server.transport.JSONTransport,org.cometd.websocket.server.WebSocketTransport</param-value>

        </init-param>

        <init-param>

            <param-name>allowedTransports</param-name>

            <param-value>org.cometd.server.transport.JSONTransport,org.cometd.websocket.server.WebSocketTransport</param-value>

        </init-param>

         <load-on-startup>1</load-on-startup>

         <async-supported>true</async-supported>

    </servlet>

    <servlet-mapping>

        <servlet-name>cometd</servlet-name>

        <url-pattern>/CometDServerDemo/*</url-pattern>

    </servlet-mapping>

<!--   <filter>

        <filter-name>cross-origin</filter-name>

        <filter-class>org.eclipse.jetty.servlets.CrossOriginFilter</filter-class>

        <async-supported>true</async-supported>

    </filter>

    <filter-mapping>

        <filter-name>cross-origin</filter-name>

        <url-pattern>/*</url-pattern>

    </filter-mapping> -->

    

<!--     <servlet>

        <servlet-name>configuration</servlet-name>

        <servlet-class>com.hisense.abby.cometd.demo.ConfigurationServlet</servlet-class>

        <load-on-startup>2</load-on-startup>

    </servlet> -->

</web-app>

 

说明：web.xml中只需要定义一个servlet就可以了；对应的class是cometd提供的类org.cometd.server.CometdServlet。需要注意的是，我们配置的url-pattern是/CometDServerDemo/*，如果配置为/*的话，握手和心跳消息都正常，但是publish消息到server的时候，会得到HTTP 302的错误码，具体原因暂时不清楚。需要特别说明的是，<async-supported>true</async-supported>这个配置项必须要有，否则启动的时候会报 ！AsyncSupported的错误，表示server当前不支持异步处理。如果配置了filter，则filter中也需要添加上这个配置。

 

3）创建普通java工程，模拟客户端订阅和发布消息

无论是发布消息还是订阅消息，都需要先和服务端握手成功：

Map<String, Object> options = new HashMap<String, Object>();

        

        ClientTransport transport = LongPollingTransport.create(null);

        

        BayeuxClient client = new BayeuxClient("http://172.16.132.120:8080/CometDServerDemo/CometDServerDemo", transport);

        

        client.handshake();

        boolean handshaken = client.waitFor(1000, BayeuxClient.State.CONNECTED);

        

        if(handshaken)

        {

 

注意：访问的url地址中有两个CometDServerDemo，第一个是工程名，第二个是我们配置的servlet中路径

 

握手成功后，发布消息：

Map<String, Object> data = new HashMap<String, Object>();

            

            for(int i=0; i<1; ++i)

            {

                data.put("xxx", i);

                

                client.getChannel("/echo").publish(data);

                

                System.out.println("publish a message:" + client.getChannel("/echo").getId() + ":" + i);

                

                Thread.sleep(3000);

            }

 

握手成功后，订阅消息：

ClientSessionChannel.MessageListener echoListener = new EchoListener();

client.getChannel(CHANNEL).subscribe(echoListener);

private static class EchoListener implements ClientSessionChannel.MessageListener

    {

        @Override

        public void onMessage(ClientSessionChannel arg0, Message arg1) {

            System.out.println("Receive message:" + arg1);

         }    

    }

 

说明：客户端的代码非常简单，握手、心跳以及底层的通信部分，客户端jar都已经封装好了。我们只需要调用对应的函数发布或者是订阅消息即可。

 

3、客户端源码分析

 

类图分析：客户端源码主要是通过BayeuxClient类操作；该类的一个实例，代表在一个和server连接的客户端；通过该类的handshake启动连接，当成功连接上server后，获取到某个channel的BayeuxClientChannel实例，就可以订阅或者是发布某个channle的数据。BayeuxClient内部有一个简易的状态机实现，主要用于维护和server之间的状态。

 

 

 

基本流程：握手成功后，发送connect请求；当有对应的订阅数据返回时，connect响应立即返回，并把其中的推送消息通知上层应用；如果没有订阅数据，则connect响应在超时后返回，仅返回connect对应的响应。当接收到connect响应后，开始发送下一次connect连接

 

Long-Polling实现：long-Polling的底层实现非常简单，采用HTTPClient实现，就是普通的HTTP消息收发。

 

WebSocket实现：底层使用了jetty的websocket实现，细节不讨论

 

4、cometd的客户端实现简述

cometd客户端的实现比较简单，就是握手消息、心跳消息[并不是严格意义上的心跳，不过是类似的功能]以及推送消息的处理。功能主要有支持同一个客户端的多channel订阅；异常时的重连、重试处理机制（不仅仅是连接，包括整个握手过程）。

 

设计上首先用了把逻辑和协议层分离，逻辑上处理都是发送握手消息，发送心跳消息，订阅频道，处理返回的响应消息。其采用了状态机来管理整个过程。关键点是在处于connected状态时，可以同时处理心跳和推送来的消息。对于long-polling模式来说，如果有推送消息，则在响应中会同时返回心跳响应和推送消息，而不是仅仅返回推送消息，所以代码可以统一在接受到心跳响应时做断开重连的处理。对于ws模式，则server端可以随时推送消息过来，客户端也是在只有收到心跳响应时，才会发送下一个心跳消息。

 

上层处理的统一，带来的是数据的冗余。无论是否有消息，服务端都会返回一个心跳响应；虽然整个心跳消息并不是必须的；而且ws模式下，也没有直接使用ws协议定义的ping、pong消息，而是直接借用了long-polling模式下的心跳消息。

评论

2 楼 Rainyn 2017-05-26  

作者写的很好。有完整项目实例吗？这个是跑在tomcat还是jetty上呢？

1 楼 hwfrog 2013-08-15  

不错， 学习了； 这段时间也想研究一下源码， 多交流