当数据库中的某些内容被修改时，仅通过 WebSockets 通知特定用户

Posted 2023-02-22

【中文标题】当数据库中的某些内容被修改时，仅通过 WebSockets 通知特定用户

【英文标题】：Notify only specific user(s) through WebSockets, when something is modified in the database

【发布时间】：2015-12-02 07:14:05

【问题描述】：

为了通过 WebSockets 通知所有用户，当在选定的 JPA 实体中修改某些内容时，我使用以下基本方法。

@ServerEndpoint("/Push")
public class Push 

    private static final Set<Session> sessions = new LinkedHashSet<Session>();

    @OnOpen
    public void onOpen(Session session) 
        sessions.add(session);
    

    @OnClose
    public void onClose(Session session) 
        sessions.remove(session);
    

    private static JsonObject createJsonMessage(String message) 
        return JsonProvider.provider().createObjectBuilder().add("jsonMessage", message).build();
    

    public static void sendAll(String text) 
        synchronized (sessions) 
            String message = createJsonMessage(text).toString();

            for (Session session : sessions) 
                if (session.isOpen()) 
                    session.getAsyncRemote().sendText(message);
                
            
        
    

当修改选定的JPA实体时，提出了适当的CDI事件，该事件将被以下CDI观察者观察到。

@Typed
public final class EventObserver 

    private EventObserver() 

    public void onEntityChange(@Observes EntityChangeEvent event) 
        Push.sendAll("updateModel");
    

观察者/消费者调用定义在 WebSockets 端点中的静态方法 Push#sendAll()，该方法将 JSON 消息作为通知发送给所有关联的用户/连接。

sendAll() 方法内部的逻辑需要以某种方式进行修改，以便只通知选定的用户。

仅通知负责修改相关实体的用户（可能是管理员用户或注册用户，只有在成功登录后才能修改某些内容）。 仅通知特定用户（不是全部）。 “特定”是指，例如，当在本网站上对帖子进行投票时，仅通知帖子所有者（该帖子可能由任何其他具有足够权限的用户投票）。

当初始握手建立时，HttpSession 可以访问，如this 回答中所述，但仍然不足以通过两个项目符号完成上述任务。由于它在第一次握手请求时可用，因此随后为该会话设置的任何属性在服务器端点中都将不可用，即，在建立握手后设置的任何会话属性都将不可用。

如上所述，仅通知选定用户的最可接受/最规范的方式是什么？ sendAll() 方法中的一些条件语句或其他地方是必需的。看来它必须做的不仅仅是用户的HttpSession。

我使用 GlassFish Server 4.1 / Java EE 7。

【参考方案1】：

会话？

因为它在第一次握手请求时可用，所以之后为该会话设置的任何属性在服务器端点中都将不可用，也就是说，在建立握手后设置的任何会话属性都将不可用可用

您似乎被“会话”一词的歧义所困扰。会话的生命周期取决于上下文和客户端。 websocket (WS) 会话的生命周期与 HTTP 会话不同。就像 EJB 会话的生命周期与 HTTP 会话不同。就像传统的 Hibernate 会话与 HTTP 会话的生命周期不同。等等。您可能已经了解的 HTTP 会话在 How do servlets work? Instantiation, sessions, shared variables and multithreading 处进行了解释。 EJB会话在这里解释JSF request scoped bean keeps recreating new Stateful session beans on every request?

WebSocket 生命周期

WS 会话与 html 文档所代表的上下文相关联。客户端基本上是 javascript 代码。 WS 会话在 JavaScript 执行 new WebSocket(url) 时启动。当 JavaScript 在 WebSocket 实例上显式调用 close() 函数时，或者当关联的 HTML 文档由于页面导航（单击链接/书签或修改浏览器地址栏中的 URL）而被卸载时，WS 会话停止，或者页面刷新，或浏览器选项卡/窗口关闭。请注意，您可以在同一个 DOM 中创建多个 WebSocket 实例，通常每个实例都有不同的 URL 路径或查询字符串参数。

每次 WS 会话开始时（即每次 JavaScript 执行 var ws = new WebSocket(url); 时），这将触发一个握手请求，您因此可以通过下面的 Configurator 类访问关联的 HTTP 会话，正如您已经找到的那样出：

public class ServletAwareConfigurator extends Configurator 

    @Override
    public void modifyHandshake(ServerEndpointConfig config, HandshakeRequest request, HandshakeResponse response) 
        HttpSession httpSession = (HttpSession) request.getHttpSession();
        config.getUserProperties().put("httpSession", httpSession);
    

因此，它不会像您预期的那样在每个 HTTP 会话或 HTML 文档中只调用一次。每次创建 new WebSocket(url) 时都会调用它。

然后将创建@ServerEndpoint 注释类的全新实例，并调用其@OnOpen 注释方法。如果您熟悉 JSF/CDI 托管 bean，只需将该类视为 @ViewScoped，将方法视为 @PostConstruct。

@ServerEndpoint(value="/push", configurator=ServletAwareConfigurator.class)
public class PushEndpoint 

    private Session session;
    private EndpointConfig config;

    @OnOpen
    public void onOpen(Session session, EndpointConfig config) 
        this.session = session;
        this.config = config;
    

    @OnMessage
    public void onMessage(String message) 
        // ...
    

    @OnError
    public void onError(Throwable exception) 
        // ...
    

    @OnClose
    public void onClose(CloseReason reason) 
        // ...
    

请注意，此类不同于例如一个 servlet 不是应用程序范围的。它基本上是 WS 会话范围的。所以每个新的 WS 会话都有自己的实例。这就是为什么您可以安全地将Session 和EndpointConfig 分配为实例变量的原因。根据类设计（例如抽象模板等），如有必要，您可以添加回 Session 作为所有其他 onXxx 方法的第一个参数。这也受支持。

当 JavaScript 执行 webSocket.send("some message") 时，将调用 @OnMessage 注释方法。 @OnClose 注解的方法将在 WS 会话关闭时被调用。如有必要，确切的关闭原因可以由CloseReason.CloseCodes 枚举提供的关闭原因代码确定。 @OnError 注解的方法会在抛出异常时被调用，通常是 WS 连接上的 IO 错误（管道断开、连接重置等）。

按登录用户收集 WS 会话

回到你只通知特定用户的具体功能需求，在上面的解释之后你应该明白你可以安全地依赖modifyHandshake()每次从关联的HTTP会话中提取登录用户，只要new WebSocket(url) 在用户登录后创建。

public class UserAwareConfigurator extends Configurator 

    @Override
    public void modifyHandshake(ServerEndpointConfig config, HandshakeRequest request, HandshakeResponse response) 
        HttpSession httpSession = (HttpSession) request.getHttpSession();
        User user = (User) httpSession.getAttribute("user");
        config.getUserProperties().put("user", user);
    

在带有@ServerEndpoint(configurator=UserAwareConfigurator.class) 的WS 端点类中，您可以在@OnOpen 注释方法中得到它，如下所示：

@OnOpen
public void onOpen(Session session, EndpointConfig config) 
    User user = (User) config.getUserProperties().get("user");
    // ...

您应该在应用程序范围内收集它们。您可以在端点类的static 字段中收集它们。或者，更好的是，如果您的环境中没有破坏 WS 端点中的 CDI 支持（在 WildFly 中工作，而不是在 Tomcat+Weld 中，不确定 GlassFish），那么只需将它们收集到一个应用程序范围的 CDI 托管 bean 中，然后再将它们收集到 @987654365 @ 在端点类中。

当User 实例不是null 时（即当用户登录时），请记住一个用户可以有多个WS 会话。因此，您基本上需要将它们收集在Map<User, Set<Session>> 结构中，或者可能是通过用户 ID 或组/角色来映射它们的更细粒度的映射，这毕竟可以更容易地找到特定用户。这一切都取决于最终的要求。这至少是一个使用应用程序范围的 CDI 托管 bean 的启动示例：

@ApplicationScoped
public class PushContext 

    private Map<User, Set<Session>> sessions;

    @PostConstruct
    public void init() 
        sessions = new ConcurrentHashMap<>();
    

    void add(Session session, User user) 
        sessions.computeIfAbsent(user, v -> ConcurrentHashMap.newKeySet()).add(session);
    

    void remove(Session session) 
        sessions.values().forEach(v -> v.removeIf(e -> e.equals(session)));
    

@ServerEndpoint(value="/push", configurator=UserAwareConfigurator.class)
public class PushEndpoint 

    @Inject
    private PushContext pushContext;

    @OnOpen
    public void onOpen(Session session, EndpointConfig config) 
        User user = (User) config.getUserProperties().get("user");
        pushContext.add(session, user);
    

    @OnClose
    public void onClose(Session session) 
        pushContext.remove(session);
    

最后，您可以在PushContext 中向特定用户发送消息，如下所示：

public void send(Set<User> users, String message) 
    Set<Session> userSessions;

    synchronized(sessions) 
        userSessions = sessions.entrySet().stream()
            .filter(e -> users.contains(e.getKey()))
            .flatMap(e -> e.getValue().stream())
            .collect(Collectors.toSet());
    

    for (Session userSession : userSessions) 
        if (userSession.isOpen()) 
            userSession.getAsyncRemote().sendText(message);
        
    

PushContext 作为 CDI 托管 bean 的另一个优势是它可以注入任何其他 CDI 托管 bean，从而更容易集成。

与关联用户触发 CDI 事件

在您的EntityListener 中，您最有可能根据您之前的相关问题Real time updates from database using JSF/Java EE 触发 CDI 事件，您已经拥有更改后的实体，因此您应该能够通过他们的模型中的关系。

只通知负责修改相关实体的用户（可能是管理员用户或注册用户，只有在成功登录后才能修改）

@PostUpdate
public void onChange(Entity entity) 
    Set<User> editors = entity.getEditors();
    beanManager.fireEvent(new EntityChangeEvent(editors));

---

只通知特定用户（不是全部）。 “具体”是指，例如，当一个帖子在本网站上被投票时，只有帖子所有者被通知（该帖子可能被任何其他具有足够权限的用户投票）。

@PostUpdate
public void onChange(Entity entity) 
    User owner = entity.getOwner();
    beanManager.fireEvent(new EntityChangeEvent(Collections.singleton(owner)));

然后在 CDI 事件观察器中，将其传递出去：

public void onEntityChange(@Observes EntityChangeEvent event) 
    pushContext.send(event.getUsers(), "message");

另见：

RFC6455 - The WebSocket Protocol（描述ws://协议） W3 - The WebSocket API（描述JSWebSocket接口） MDN - Writing WebSocket client application（描述如何在客户端使用WS API） Java EE 7 tutorial - WebSocket（描述javax.websocket API 以及如何使用它） Real time updates from database using JSF/Java EE（如果你已经在使用 JSF）

【讨论】：

谢谢。富有表现力的回答。但是，我不知道由谁来调用一组用户的 send() 方法。当需要作为参数传递给此方法的一组用户不可用时触发适当的 CDI 事件时，它应该由适当的 CDI 观察者调用。

“当修改选定的JPA实体时，提出了适当的CDI事件，该事件将被以下CDI观察者观察到。” i>在您创建@ 987654382时@，您已经知道哪个实体（以及哪些用户）受到影响，所以只需像 new EntityChangeEvent(users) 一样，将其传递并让观察者通过 getter 从中提取。

new 发生在像event.fire(new EntityChangeEvent(entity)); 这样的 CDI 事件被触发时。一组用户 (Set<User>) 当时不可用。提供给构造函数的参数是触发事件的当前实体。引发事件的实体可能不是User。负责执行该操作/修改进而引发事件的当前“登录”在该位置无法识别。 （例如，任何注册用户都可以提交（插入）反馈。受“插入”影响的实体是Feedback，而不是User）。

那么，Feedback 与任何User （通过例如数据库中的 FK 和实体中的@OneToOne/Many）没有直接关系？这怎么可能？ :) 或者您只是想通知当前在特定页面上的 任何 用户？然后，该信息应作为路径或查询参数在 WS URL 中传递，但这与当前提出的问题完全不同。

通过外键关系与User相关。我认为这应该导致LazyInitializationException。 （我没有测试，因为 GlassFish Server 当前无法运行。操作系统有一些问题，可能需要重新安装 :)）。几天后我也许会付诸实践。谢谢。