javax.websockets / Tyrus 中的线程

Posted 2023-02-16

【中文标题】javax.websockets / Tyrus 中的线程

【英文标题】：Threading in javax.websockets / Tyrus

【发布时间】：2015-08-08 16:19:54

【问题描述】：

我正在编写一个从 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");
        
    

【问题讨论】：

这可能是“最大与核心大小”的问题，请参阅this question。 IE。尝试使用setCorePoolSize 而不是“maxPoolSize”。

不，我也试过了。但是我现在发现messageHandlers 的文档只允许一个线程。

【参考方案1】：

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

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

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

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

每个 Web 套接字会话一次使用不超过一个线程来调用其 MessageHandler。

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

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

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

如果您关心客户端和服务器之间的会话数（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 中似乎也是如此。这让我跑了好几个小时，以为我不了解线程模型，而只是池大小让我感到困惑。