java - javax.websockets/Tyrus 中的线程

Question

我正在编写一个从 websocket 服务器发送和接收消息的 Java 应用程序。当应用程序收到一条消息时，可能需要一些时间来处理它。因此，我正在尝试使用多个线程来接收消息。据我了解，Grizzly 具有选择器线程和工作线程。默认情况下有 1 个选择器线程和 2 个工作线程，在下面的示例中，我试图将它们分别增加到 5 和 10。在下面的示例中，我将调用 onMessage 方法的线程暂停 10 秒，以模拟传入信息的处理。信息每秒传入一次，因此 10 个线程应该能够处理流量。当我分析运行时，只有 1 个选择器线程正在运行和 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 session 上处理多个线程，通常需要 WebSocket 能够多路复用进出客户端的数据。这目前不是 WebSockets 的一个特性，但肯定可以建立在它之上。但是，在单个 channel 上多路复用这些客户端和服务器线程会带来相当大的复杂性，因为您需要阻止所有客户端和服务器线程无意中覆盖或彼此饥饿。

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.

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

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

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

如果您担心客户端和服务器之间的 session 数(TCP/HTTP 连接)，您可以考虑在客户端创建一个 session 池，并一次使用每个客户端 session ，每当客户端完成 session 时，将 session 返回到池中。

最后，也许没有直接关系:我发现当我使用 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 似乎也是如此。这让我跑了几个小时，以为我不了解线程模型，而当时让我感到困惑的只是池大小。