在 javax.websockets/Tyrus 中穿线

Question

我正在编写一个 Java 应用程序，用于从 websocket 服务器发送和接收消息。当应用程序收到一条消息时，可能需要一些时间来处理它。因此，我正在尝试使用多个线程来接收消息。据我了解 Grizzly 有选择器线程和工作线程。默认情况下有 1 个选择器线程和 2 个工作线程，在下面的示例中，我试图将它们分别增加到 5 和 10。在下面的示例中，我将调用 onMessage 方法的线程暂停 10 秒，以模拟传入信息的处理。信息每秒传入一次，因此 10 个线程应该能够处理流量。当我分析运行时，只有 1 个选择器线程是运行ning 和 2 个工作线程。此外，消息仅以 10 秒的间隔接收。表示只有 1 个线程正在处理流量 - 我觉得这很奇怪。在分析期间，一个工作线程，例如Grizzly(1) 收到发送的第一条消息。然后 10 秒后 'Grizzly(2)' 收到第二条消息 - 然后 Grizzly(2) 继续接收消息，并且 Grizzly(1) 不执行任何操作。

有人可以解释一下这种奇怪的行为以及如何将其更改为例如10 个线程不断排队等待消息？

主要：

    public static void main(String[] args) {
        WebsocketTextClient client = new WebsocketTextClient();
        client.connect();
        for (int i = 0; i < 60; i++) {
            client.send("Test message " + i);
            try {
                Thread.sleep(1000);
            } catch (Exception e) {
                System.out.println("Error sleeping!");
            }
        }
    }

WebsocketTextClient.java:

import java.net.URI;
import javax.websocket.ClientEndpointConfig;
import javax.websocket.EndpointConfig;
import javax.websocket.Session;
import javax.websocket.Endpoint;
import javax.websocket.MessageHandler;
import org.glassfish.tyrus.client.ClientManager;
import org.glassfish.tyrus.client.ThreadPoolConfig;
import org.glassfish.tyrus.container.grizzly.client.GrizzlyClientProperties;

public class WebsocketTextClient {

    private ClientManager client;
    private ClientEndpointConfig clientConfig;
    WebsocketTextClientEndpoint endpoint;

    public WebsocketTextClient() {
        client = ClientManager.createClient();
        client.getProperties().put(GrizzlyClientProperties.SELECTOR_THREAD_POOL_CONFIG, ThreadPoolConfig.defaultConfig().setMaxPoolSize(5));
        client.getProperties().put(GrizzlyClientProperties.WORKER_THREAD_POOL_CONFIG, ThreadPoolConfig.defaultConfig().setMaxPoolSize(10));
    }

    public boolean connect() {
        try {
            clientConfig = ClientEndpointConfig.Builder.create().build();
            endpoint = new WebsocketTextClientEndpoint();
            client.connectToServer(endpoint, clientConfig, new URI("wss://echo.websocket.org"));
        } catch (Exception e) {
            return false;
        }
        return true;
    }

    public boolean disconnect() {
        return false;
    }

    public boolean send(String message) {
        endpoint.session.getAsyncRemote().sendText(message);
        return true;
    }

    private class WebsocketTextClientEndpoint extends Endpoint {
        Session session;

        @Override
        public void onOpen(Session session, EndpointConfig config) {
            System.out.println("Connection opened");
            this.session = session;
            session.addMessageHandler(new WebsocketTextClientMessageHandler());
        }
    }

    private class WebsocketTextClientMessageHandler implements MessageHandler.Whole<String> {

        @Override
        public void onMessage(String message) {
            System.out.println("Message received from " + Thread.currentThread().getName() + " " + message);
            try {
                Thread.sleep(10000);
            } catch (Exception e) {
                System.out.println("Error sleeping!");
            }
            System.out.println("Resuming");
        }
    }
}

Answer 1

您似乎在要求 WebSockets 能够接收由同一客户端连接发送的多条消息，在单独的线程中处理这些消息，并在响应准备就绪时发送响应 - 这意味着，可能出问题了。这种情况只有在客户端是多线程的情况下才会发生。

要在同一个 WebSocket 会话上处理多个线程，通常需要 WebSocket 能够多路复用进出客户端的数据。这目前不是 WebSockets 的一个特性，但肯定可以建立在它之上。但是，在单个通道上多路复用这些客户端和服务器线程会带来相当大的复杂性，因为您需要阻止所有客户端和服务器线程无意中覆盖或相互饿死。

MessageHandler 的 Java 规范在线程模型方面可能有点模棱两可；

https://docs.oracle.com/javaee/7/api/javax/websocket/MessageHandler.html 说：

Each web socket session uses no more than one thread at a time to call its MessageHandlers.

但这里的重要术语是 "socket session"。如果您的客户端在同一 WebSocket 会话中发送多条消息，则服务器端处理程序将在单个线程中执行。这并不意味着您不能在线程内做很多有趣的事情，特别是如果您在两端都使用 Input/OutputStreams（或 Writers）。确实意味着与客户端的通信仅由一个线程进行调解。如果您想多路复用通信，则必须在套接字之上写一些东西才能做到；这将包括开发您自己的线程模型来分派请求。

一个更简单的解决方案是为每个客户端请求创建一个新会话。每个客户端请求启动一个会话（即TCP连接），发送数据，等待结果。这为您提供了多个 MessageHandler 线程 - 每个会话一个，根据规范。

这是在服务器端获得多线程的最直接的方法；任何其他方法往往需要多路复用机制——根据您的用例，这可能不值得付出努力，而且肯定会带来一些复杂性和风险。

如果您担心 client/s 和 server/s 之间的会话数（TCP/HTTP 连接），您可以考虑在客户端创建一个会话池，然后使用每个客户端一次一个会话，只要客户端完成会话，就会将会话返回到池中。

最后，也许没有直接关系：我发现当我使用 Payara Micro 为 WebSocket 端点提供服务时，我需要设置这个：

  <resources>
    ...
    <managed-executor-service maximum-pool-size="200" core-pool-size="10" long-running-tasks="true" keep-alive-seconds="300" hung-after-seconds="300" task-queue-capacity="20000" jndi-name="concurrent/__defaultManagedExecutorService" object-type="system-all"></managed-executor-service>

默认的ManagedExecutorService只提供单线程。 Glassfish 似乎也是如此。这让我运行想了好几个小时，认为我不了解线程模型，而当时只是池大小让我感到困惑。

在 javax.websockets/Tyrus 中穿线

Threading in javax.websockets / Tyrus

java

multithreading

websocket

tyrus