java - CometD サーバーが例外をスローし、一度にすべてのメッセージをプッシュしない

Question

CometD を起動すると、Tomcat 7.0 で次の例外が発生します。メッセージをプッシュしようとすると、メッセージをプッシュするのに6000ミリ秒かかることがあり、一度にすべてではなく一度に2つずつメッセージをプッシュします。10 を 200 にプッシュしようとしましたが、状況は変わりません。私はすでに jetty-websocket-7.6.10.v20130312.jar と jetty-server-7.6.13.v20130916.jar を CometD サーバーアプリケーションに含めましたが、それでもこの例外が発生します。

例外：

23 Oct, 2013 10:12:45 AM org.apache.catalina.core.StandardWrapperValve invoke
SEVERE: Servlet.service() for servlet [cometd] in context with path [/Samp] threw   exception
java.lang.NullPointerException
at org.eclipse.jetty.websocket.WebSocketFactory.upgrade(WebSocketFactory.java:238)
at org.eclipse.jetty.websocket.WebSocketFactory.acceptWebSocket(WebSocketFactory.java:396)
at org.cometd.websocket.server.WebSocketTransport.handle(WebSocketTransport.java:157)
at org.cometd.server.CometdServlet.service(CometdServlet.java:166)
at javax.servlet.http.HttpServlet.service(HttpServlet.java:722)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:305)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:210)
at org.eclipse.jetty.servlets.CrossOriginFilter.handle(CrossOriginFilter.java:247)
at org.eclipse.jetty.servlets.CrossOriginFilter.doFilter(CrossOriginFilter.java:210)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:243)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:210)
at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:224)
at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:169)
at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:472)
at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:168)
at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:98)
at org.apache.catalina.valves.AccessLogValve.invoke(AccessLogValve.java:928)
at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:118)
at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:407)
at org.apache.coyote.http11.AbstractHttp11Processor.process(AbstractHttp11Processor.java:987)
at org.apache.coyote.AbstractProtocol$AbstractConnectionHandler.process(AbstractProtocol.java:539)
at org.apache.tomcat.util.net.JIoEndpoint$SocketProcessor.run(JIoEndpoint.java:298)
at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)

23 Oct, 2013 10:12:46 AM org.apache.catalina.core.StandardWrapperValve invoke
SEVERE: Servlet.service() for servlet [cometd] in context with path [/Samp] threw exception
java.lang.IllegalStateException: Not supported.
at org.apache.catalina.connector.Request.startAsync(Request.java:1664)
at org.apache.catalina.connector.Request.startAsync(Request.java:1657)
at org.apache.catalina.connector.RequestFacade.startAsync(RequestFacade.java:1022)
at org.eclipse.jetty.continuation.Servlet3Continuation.suspend(Servlet3Continuation.java:189)
at org.cometd.server.transport.LongPollingTransport.handle(LongPollingTransport.java:289)
at org.cometd.server.CometdServlet.service(CometdServlet.java:166)
at javax.servlet.http.HttpServlet.service(HttpServlet.java:722)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:305)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:210)
at org.eclipse.jetty.servlets.CrossOriginFilter.handle(CrossOriginFilter.java:247)
at org.eclipse.jetty.servlets.CrossOriginFilter.doFilter(CrossOriginFilter.java:210)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:243)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:210)
at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:224)
at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:169)
at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:472)
at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:168)
at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:98)
at org.apache.catalina.valves.AccessLogValve.invoke(AccessLogValve.java:928)
at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:118)
at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:407)
at org.apache.coyote.http11.AbstractHttp11Processor.process(AbstractHttp11Processor.java:987)
at org.apache.coyote.AbstractProtocol$AbstractConnectionHandler.process(AbstractProtocol.java:539)
at org.apache.tomcat.util.net.JIoEndpoint$SocketProcessor.run(JIoEndpoint.java:300)
at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)

サービスクラス:

@Service
public class HelloService
{
@Inject
private BayeuxServer bayeux;
@Session
private ServerSession serverSession;

private static int msgCount = 0;

private  static Logger log;

@Listener("/service/java")
public void processMsgFromJava(ServerSession remote, ServerMessage.Mutable message)
{
    msgCount++;

    String msgId = "EVENT-"+msgCount;
    Date dt = Calendar.getInstance().getTime();

    Map<String, Object> input = message.getDataAsMap();
    String eventId = (String)input.get("eventID");

    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

    String channelName = "/java/test";
    // Initialize the channel, making it persistent and lazy
    bayeux.createIfAbsent(channelName, new ConfigurableServerChannel.Initializer(){
        public void configureChannel(ConfigurableServerChannel channel){
            channel.setPersistent(false);
            channel.setLazy(false);
        }
    });

    // Publish to all subscribers
    ServerChannel channel = bayeux.getChannel(channelName);
    input.put("ms", System.currentTimeMillis());
    channel.publish(serverSession, input, null);
    log.info("Msg published 111111111111........"+msgId);
    System.out.println(channel.isPersistent());
 }
   }

クライアントクラス:

private ClientTransport transport= null;
private BayeuxClient client = null;

private ClientSessionChannel clientSession = null;
public static int msgCount = 0;

public boolean pushToComet(String cometDURL, String subscriberChannel, Map<String,Object> msgToPublish) throws Exception{
    boolean isSuccess = true;

    try {

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

        /* Prepare the WebSocket transport*/
        WebSocketClientFactory wsFactory = new WebSocketClientFactory();
        wsFactory.start();
        ClientTransport wsTransport = new WebSocketTransport(null, wsFactory, null);
        wsTransport.setOption("maxNetworkDelay", 0);
        wsTransport.setOption("connectTimeout", 10000);
        wsTransport.setOption("idleTimeout", 10000);
        wsTransport.setOption("maxMessageSize", 8192);

        transport = LongPollingTransport.create(options);
    //  transport.setDebugEnabled(true);
        transport.setOption("maxNetworkDelay", 0);

        client = new BayeuxClient(cometDURL,wsTransport, transport);
        //client.setDebugEnabled(true);

        client.handshake();

        boolean handshaken = client.waitFor(2000, BayeuxClient.State.CONNECTED);
        if(handshaken){
            msgCount++;
            clientSession = client.getChannel(subscriberChannel);
            clientSession.publish(msgToPublish);
        }else{
            isSuccess=false;
            System.out.println("CommetD not connected????????");
        }
    } catch (Exception e) {
        isSuccess=false;
        e.printStackTrace();
    }finally{
//          close();
    }
    return isSuccess;
}

private void close(){
        clientSession.release();
        client.disconnect();
        client.waitFor(1000, BayeuxClient.State.DISCONNECTED);
        transport.terminate();

}

public void closeConn(){
    client.disconnect();
    client.waitFor(1000, BayeuxClient.State.DISCONNECTED);
    transport.terminate();
}

web.xml

<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_2_5.xsd"
     version="2.5">

<servlet>
    <servlet-name>cometd</servlet-name>
    <servlet-class>org.cometd.annotation.AnnotationCometdServlet</servlet-class>
    <init-param>
        <param-name>transports</param-name>
        <param-value>org.cometd.websocket.server.WebSocketTransport</param-value>
    </init-param>
    <init-param> 
        <param-name>services</param-name>
        <param-value>galaxy_comet.HelloService</param-value>
    </init-param>
    <init-param>
        <param-name>maxLazyTimeout</param-name>
        <param-value>-1</param-value>
    </init-param>
     <init-param>
        <param-name>timeout</param-name>
        <param-value>2000</param-value>
    </init-param>
    <init-param>
        <param-name>interval</param-name>
        <param-value>5000</param-value>
    </init-param>
    <init-param>
        <param-name>maxInterval</param-name>
        <param-value>10000</param-value>
    </init-param>
    <init-param>
        <param-name>maxQueue</param-name>
        <param-value>-1</param-value>
    </init-param>
    <init-param>
        <param-name>metaConnectDeliverOnly</param-name>
        <param-value>true</param-value>
    </init-param>
   <!--  <init-param>
        <param-name>multiFrameInterval</param-name>
        <param-value>200</param-value>
    </init-param> -->
    <load-on-startup>1</load-on-startup>

</servlet>
<servlet-mapping>
    <servlet-name>cometd</servlet-name>
    <url-pattern>/cometd/*</url-pattern>
</servlet-mapping>


<filter>
    <filter-name>cross-origin</filter-name>
    <filter-class>org.eclipse.jetty.servlets.CrossOriginFilter</filter-class>
</filter>
<filter-mapping>
    <filter-name>cross-origin</filter-name>
    <url-pattern>/cometd/*</url-pattern>
</filter-mapping>

<listener>
    <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
</listener>

jspのJavascript

(function($)
{
var cometd = $.cometd;

$(document).ready(function()
{
    function _connectionEstablished()
    {
        $('#body').append('<div>CometD Connection Established</div>');
    }

    function _connectionBroken()
    {
        $('#body').append('<div>CometD Connection Broken</div>');
    }

    function _connectionClosed()
    {
        $('#body').append('<div>CometD Connection Closed</div>');
    }

    // Function that manages the connection status with the Bayeux server
    var _connected = false;
    function _metaConnect(message)
    {
        if (cometd.isDisconnected())
        {
            _connected = false;
            _connectionClosed();
            return;
        }

        var wasConnected = _connected;
        _connected = message.successful === true;
        if (!wasConnected && _connected)
        {
            _connectionEstablished();
        }
        else if (wasConnected && !_connected)
        {
            _connectionBroken();
        }
    }

     function _metaHandshake(handshake)
    {
        if (handshake.successful === true)
        {
            cometd.batch(function()
            {
                cometd.subscribe('/java/test', function(message)
                {
                    var d = new Date();
                    $('#body').append('<div>Server Says: ' + message.data.eventID + ':'+ (d.getTime()-message.data.ms) + '</div>');
                });
            });
        }
    }

    // Disconnect when the page unloads
    $(window).unload(function()
    {
        cometd.disconnect(true);
    });

    var cometURL = "http://localhost:8080/cometd1/cometd";
    cometd.configure({
        url: cometURL,
        logLevel: 'debug'
    });

    cometd.addListener('/meta/handshake', _metaHandshake);
    cometd.addListener('/meta/connect', _metaConnect);

    cometd.handshake();
});
})(jQuery);

channel is persistent() のコンソールメッセージは常に false として出力されます。

クライアントクラスの final で close() を呼び出すと、NullPointerException が発生します。

ベンチマークに記載されている結果を得ることができません。どこが間違っているのかわかりません。channel.setLazy(false) も作成しましたが、まだ時間がかかります。また、複数のクライアントが同じチャネルにサブスクライブしている場合、これらのクライアントでのメッセージのプッシュ時間は異なります (たとえば、最初のクライアントでは発行に 2 秒かかり、2 番目のクライアントでは発行に 4 秒かかるなど)。）。一度にメッセージをブロードキャストしていないことを意味します。どうしてこうなった。websocket transport なしで試してみました。その時もメッセージが遅れています。websocket と longpolling の両方を使用しようとすると、この問題が発生します。この点に関するヘルプとガイダンスは、私を大いに助けてくれます。

java - CometD サーバーが例外をスローし、一度にすべてのメッセージをプッシュしない

1 に答える 1

Related

Reference