如何获取现有的 websocket 实例

2022-09-03 10:09:08

我正在开发一个应用程序,该应用程序使用Websockets(Java EE 7)以异步方式向所有连接的客户端发送消息。每当创建新文章(我的应用中的参与模式)时,服务器(Websocket 终结点)都应发送这些消息。

每次与 websocket 终结点建立连接时,我都会将相应的会话添加到列表中,以便可以在外部访问该列表。

但我遇到的问题是,当我访问这个创建的websocket端点时,所有客户端都从外部(任何其他业务类)连接到该端点,我已经获得了现有的实例(如单例)。

所以,你能不能建议我一种可以获得websocket端点的现有实例的方法,因为我不能把它创建为新的MyWebsocketEndPoint() coz,每当收到来自客户端的请求时,它都会由websocket内部机制创建。

对于参考:

private static WebSocketEndPoint INSTANCE = null;

public static WebSocketEndPoint getInstance() {
if(INSTANCE == null) {
// Instead of creating a new instance, I need an existing one
    INSTANCE = new WebSocketEndPoint ();
}
        return INSTANCE;
}

提前致谢。


答案 1

容器会为每个客户端连接创建一个单独的终结点实例,因此您无法执行要执行的操作。但我认为您尝试做的是在事件发生时向所有活动的客户端连接发送消息,这相当简单。

该类具有检索表示与该会话关联的终结点的实例的方法。javax.websocket.SessiongetBasicRemoteRemoteEndpoint.Basic

您可以通过调用 来检索所有打开的会话,然后循环访问它们。该循环将向每个客户端连接发送一条消息。下面是一个简单的示例:Session.getOpenSessions()

@ServerEndpoint("/myendpoint")
public class MyEndpoint {
  @OnMessage
  public void onMessage(Session session, String message) {
    try {  
      for (Session s : session.getOpenSessions()) {
        if (s.isOpen()) {
          s.getBasicRemote().sendText(message);
        }
    } catch (IOException ex) { ... }
  } 
} 

但在您的案例中,您可能希望使用 CDI 事件来触发对所有客户端的更新。在这种情况下,您将创建一个 CDI 事件,Websocket 端点类中的方法会观察到该事件:

@ServerEndpoint("/myendpoint")
public class MyEndpoint {
  // EJB that fires an event when a new article appears
  @EJB
  ArticleBean articleBean;
  // a collection containing all the sessions
  private static final Set<Session> sessions = 
          Collections.synchronizedSet(new HashSet<Session>());

  @OnOpen
  public void onOpen(final Session session) {
    // add the new session to the set
    sessions.add(session);
    ...
  }

  @OnClose
  public void onClose(final Session session) {
    // remove the session from the set
    sessions.remove(session);
  }

  public void broadcastArticle(@Observes @NewArticleEvent ArticleEvent articleEvent) {
    synchronized(sessions) {
      for (Session s : sessions) {
        if (s.isOpen()) {
          try {
            // send the article summary to all the connected clients
            s.getBasicRemote().sendText("New article up:" + articleEvent.getArticle().getSummary());
          } catch (IOException ex) { ... }
        }
      }
    }
  }
}

上面示例中的 EJB 将执行如下操作:

...
@Inject
Event<ArticleEvent> newArticleEvent;

public void publishArticle(Article article) {
  ...
  newArticleEvent.fire(new ArticleEvent(article));
  ...
}

请参阅 Java EE 7 教程中有关 WebSocketsCDI 事件的章节。

编辑:修改了使用事件作为参数的方法。@Observer

编辑2:将循环包装在广播文章中同步,按@gcvt。

编辑 3:更新了 Java EE 7 教程的链接。干得不错,甲骨文。嘘。


答案 2

实际上,WebSocket API 提供了一种控制端点实例化的方法。查看 https://tyrus.java.net/apidocs/1.2.1/javax/websocket/server/ServerEndpointConfig.Configurator.html

简单样本(取自 Tyrus - WebSocket RI 测试):

    public static class MyServerConfigurator extends ServerEndpointConfig.Configurator {

        public static final MyEndpointAnnotated testEndpoint1 = new MyEndpointAnnotated();
        public static final MyEndpointProgrammatic testEndpoint2 = new MyEndpointProgrammatic();

        @Override
        public <T> T getEndpointInstance(Class<T> endpointClass) throws InstantiationException {
            if (endpointClass.equals(MyEndpointAnnotated.class)) {
                return (T) testEndpoint1;
            } else if (endpointClass.equals(MyEndpointProgrammatic.class)) {
                return (T) testEndpoint2;
            }

            throw new InstantiationException();
        }
    }

您需要将其注册到终端节点:

@ServerEndpoint(value = "/echoAnnotated", configurator = MyServerConfigurator.class)
public static class MyEndpointAnnotated {

    @OnMessage
    public String onMessage(String message) {

        assertEquals(MyServerConfigurator.testEndpoint1, this);

        return message;
    }
}

或者,您也可以将其与编程终结点一起使用:

public static class MyApplication implements ServerApplicationConfig {
    @Override
    public Set<ServerEndpointConfig> getEndpointConfigs(Set<Class<? extends Endpoint>> endpointClasses) {
        return new HashSet<ServerEndpointConfig>
          (Arrays.asList(ServerEndpointConfig.Builder
            .create(MyEndpointProgrammatic.class, "/echoProgrammatic")
            .configurator(new MyServerConfigurator())
            .build()));
    }

    @Override
    public Set<Class<?>> getAnnotatedEndpointClasses(Set<Class<?>> scanned) {
        return new HashSet<Class<?>>(Arrays.asList(MyEndpointAnnotated.class));
    }

当然,如果您将有一个配置器用于所有端点(如所示代码段所示的丑陋 ifs),或者您将为每个端点创建单独的配置器,则由您决定。

请不要按原样复制显示的代码 - 这只是Tyrus测试的一部分,它确实违反了一些基本的OOM范式。

有关完整测试,请参阅 https://github.com/tyrus-project/tyrus/blob/1.2.1/tests/e2e/src/test/java/org/glassfish/tyrus/test/e2e/GetEndpointInstanceTest.java


推荐