JMS-Websocket - 延迟消息传递

【问题标题】：JMS-Websocket - delayed message deliveryJMS-Websocket - 延迟消息传递
【发布时间】：2019-07-18 07:50:15
【问题描述】：

此应用程序从数据库事件接收消息并将其转发到客户端应用程序。当客户端浏览器有一个 Web 套接字会话时，消息会立即传递。

但是，当不存在 Web 套接字会话并且 JMSProducer 将消息发送到 QueueSenderSessionBean 中的目标“jms/notificationQueue”时，消息会立即在 NotificationEndpoint 中使用。这不是我的意图。

我的意图是让队列保留消息，直到用户连接到 NotificationEndpoint。如果用户没有连接到 NotificationEndpoint，我认为应该没有创建 NotificationEndpoint 的实例来接收消息。

如何延迟 JMSConsumer 从队列中消费消息？

概览 - TomEE Plus 8.0.0-M1 项目

  1. 应用程序在 NotificationServlet 中接收通知 HttpServletRequest
  2. 通过QueueSenderSessionBean注入NotificationServlet将字符串消息放入JMS队列
  3. NotificationMessageDrivenBean 实现 MessageListener 来监听 JMS 队列
  4. 使用 @NotificationServletJMSMessage 注释的事件从 NotificationMessageDrivenBean 中为观察者触发 JMSMessage 上的 NotificationEndpoint 方法。
  5. NotificationEndpoint 使用 PushContext 收集所有 websocket 会话以将消息传递给用户
  6. 在 PushContext.send 中，如果任何 websocket 会话的用户 uuid 属性与消息用户 uuid 属性匹配，则消息为 传递到每个 websocket 会话。

我对@ServerEndpoint 的理解是“每个新的 WS 会话都有自己的实例”。 Notify only specific user(s) through WebSockets, when something is modified in the database

来源：以上链接来自https://stackoverflow.com/users/157882/balusc 和https://blogs.oracle.com/theaquarium/integrating-websockets-and-jms-with-cdi-events-in-java-ee-7-v2

WEB-INF/resources.xml

<?xml version="1.0" encoding="UTF-8"?>
<resources>
    <Resource id="jmsConnectionFactory" type="javax.jms.ConnectionFactory">
        connectionMaxIdleTime = 15 Minutes
        connectionMaxWaitTime = 5 seconds
        poolMaxSize = 10
        poolMinSize = 0
        resourceAdapter = Default JMS Resource Adapter
        transactionSupport = xa
    </Resource>
</resources>

NotificationServlet.java

import java.io.IOException;
import java.util.UUID;

import javax.annotation.Resource;
import javax.faces.context.FacesContext;
import javax.inject.Inject;
import javax.jms.Queue;
import javax.servlet.ServletException;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

@WebServlet("/notifications")
public class NotificationServlet extends HttpServlet
{
    @Resource(name = "jms/notificationQueue")
    private Queue _notificationQueue;

    @Inject
    private QueueSenderSessionBean _queueSessionSenderBean;

    @Override
    protected void doGet(HttpServletRequest request, 
            HttpServletResponse response) 
        throws ServletException, 
        IOException
    {
        try
        {
            String notificationJson =
                    extractNotificationJson(request);
            if (notificationJson != null)
            {
                _queueSessionSenderBean.sendMessage(
                        "notification=" 
                                + notificationJson);                
            }

        }
        catch (Exception e)
        {
            e.printStackTrace();
            // handle exception
        }
    }

    public String extractNotificationJson(HttpServletRequest request) 
            throws IOException
    {
        if(request.getParameter("notification") != null)
        {
            String[] notificationString = 
                    request.getParameterValues("notification");
            return notificationString[0];
        }
        return null;       
    }
}

QueueSenderSessionBean.java

import javax.annotation.Resource;
import javax.ejb.LocalBean;
import javax.ejb.Stateless;
import javax.inject.Inject;
import javax.inject.Named;
import javax.jms.DeliveryMode;
import javax.jms.JMSConnectionFactory;
import javax.jms.JMSContext;
import javax.jms.JMSException;
import javax.jms.JMSProducer;
import javax.jms.Queue;
import javax.jms.TextMessage;

import org.json.JSONObject;

@Named
@LocalBean
@Stateless
public class QueueSenderSessionBean
{
    @Resource(mappedName = "jms/notificationQueue")
    private Queue _notificationQueue;

    @Inject
    @JMSConnectionFactory("jmsConnectionFactory")
    private JMSContext _jmsContext; 

    // Static Methods

    // Member Methods
    public void sendMessage(String message) 
    {
        try
        {        
            JMSProducer messageProducer =
                _jmsContext.createProducer();
            messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);

            String userProperty = "someValue";

            TextMessage textMessage = _jmsContext.createTextMessage(message);
            textMessage.setStringProperty("userProperty", userProperty);            
            messageProducer.send(_notificationQueue, textMessage);

        }
        catch (JMSException e)
        {
            e.printStackTrace();
            // handle jms exception
        }
    }
}

限定符 NotificationServletJMSMessage.java

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

import javax.inject.Qualifier;

    @Qualifier
    @Retention(RetentionPolicy.RUNTIME)
    @Target({ElementType.METHOD, ElementType.FIELD, ElementType.PARAMETER, ElementType.TYPE})
    public @interface NotificationServletJMSMessage
    {

    }

NotificationMessageDrivenBean.java

import javax.ejb.MessageDriven;
import javax.enterprise.event.Event;
import javax.inject.Inject;
import javax.inject.Named;
import javax.jms.Message;
import javax.jms.MessageListener;

@Named
@MessageDriven(mappedName = "jms/notificationQueue")
public class NotificationMessageDrivenBean implements MessageListener
{
    @Inject
    @NotificationServletJMSMessage
    Event<Message> jmsEvent;

    @Override
    public void onMessage(Message message)
    {
        jmsEvent.fire(message);
    }
}

PushContext.java

import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;
import javax.jms.JMSConsumer;
import javax.jms.JMSContext;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Queue;
import javax.jms.TextMessage;
import javax.websocket.Session;

@ApplicationScoped
public class PushContext
{
    @Inject 
    private JMSContext _jmsContext; 

    @Resource(mappedName = "jms/notificationQueue")
    private Queue _notificationQueue;

    private Map<String, Set<Session>> _sessions;

    @PostConstruct 
    public void init()
    {
        _sessions = new ConcurrentHashMap<>();
    }

    public void add(Session session, String userUuid)
    {
        _sessions.computeIfAbsent(userUuid, 
                value -> ConcurrentHashMap.newKeySet()).add(session);
    }

    void remove(Session session)
    {
        _sessions.values().forEach(value -> value.removeIf(e -> e.equals(session)));
    }

    public void send(Set<String> userUuids, Message message) throws JMSException 
    {
        String userUuid = message.getStringProperty("userUuid");
        userUuids.add(userUuid);

        Set<Session> userSessions;

        synchronized(_sessions) 
        {
            userSessions = _sessions.entrySet().stream()
                .filter(e -> userUuids.contains(e.getKey()))
                .flatMap(e -> e.getValue().stream())
                .collect(Collectors.toSet());
        }
        for (Session userSession : userSessions) 
        {
            if (userSession.isOpen()) 
            {
                userSession.getAsyncRemote().sendText(((TextMessage) message).getText());
            }
        }
    }

    public void removeSession(Session session)
    {
        String userUuid = (String)session.getUserProperties().get("userUuid");
        _sessions.remove(userUuid, session);
    }
}

NotificationEndpoint.java

import java.io.IOException;
import java.io.Serializable;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;

import javax.enterprise.event.Observes;
import javax.inject.Inject;
import javax.inject.Named;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.websocket.CloseReason;
import javax.websocket.EndpointConfig;
import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;

@Named
@ServerEndpoint(value="/notificationEndpoint/{tokenId}")
public class NotificationEndpoint 
{
    private static final Set<Session> SESSIONS =
            Collections.synchronizedSet(new HashSet<Session>()); 
    private QueueSenderSessionBean _senderBean;

    @Inject
    private PushContext _pushContext;

    @Inject
    public NotificationEndpoint(QueueSenderSessionBean senderBean)
    { 
        _senderBean = senderBean;
    }

    @OnOpen
    public void onOpen(Session session,
            EndpointConfig configurator,
            @PathParam(value = "tokenId") String userUuidString) 
    {
        session.getUserProperties().put("userUuid", userUuidString);        
        _pushContext.add(session, userUuidString);
    }


    @OnMessage
    public void onMessage(String message, Session session) 
            throws IOException 
    {
        System.out.println("Message received: " + message);
        _senderBean.sendMessage(message);
    }

    @OnClose
    public void onClose(CloseReason reason, Session session) 
    {
        System.out.println(
                "Closing 'notificatioEndpoint due to " 
                + reason.getReasonPhrase());
        try
        {
            session.close();
        }
        catch (IOException e)
        {
            e.printStackTrace();
        }
        _pushContext.removeSession(session);
    }

    @OnError
    public void error(Session session, Throwable t) 
    {
       t.printStackTrace();
    }

    public static void sendToAllClients(String message) 
    {
        synchronized (SESSIONS) 
        {
            for (Session session : SESSIONS) 
            {
                if (session.isOpen()) 
                {
                    session.getAsyncRemote().sendText(message);
                }
            }
        }
    }

    public void onJMSMessage(@Observes @NotificationServletJMSMessage Message message) 
    {
        Set<String> userUuids = new HashSet<String>();

        try 
        {
            _pushContext.send(userUuids, message);
        } 
        catch (JMSException ex) 
        {
            ex.printStackTrace();
            Logger.getLogger(NotificationEndpoint.class.getName()).
            log(Level.SEVERE, null, ex);
        }     
    }
}

谢谢你， 泰德S

【问题讨论】：

    标签： java websocket jms message-queue observer-pattern


    【解决方案1】：

    在here 解决方案的启发下实现了延迟消息传递。

    如果用户未连接到 Web 套接字，则解决方案是使用本地队列来保存消息，然后在连接时将消息从本地队列移动到远程队列，该队列使用 MessageDrivenBean 立即接收/使用。

    另外，我没有使用 Web Servlet 监听来自数据库 (Postgresql) 的消息，而是将我的数据库触发器更改为 NOTIFY，并使用 pgjdbc-ng 驱动程序和 Postgresql LISTEN/NOTIFY 模式启动了一个异步监听器 here .

    NotificationListener.java

    @Stateless
public class NotificationListener extends Thread
{
    @Inject
    private QueueSenderSessionBean _queueSessionSenderBean;

    @Override
    public void run()
    {
        listenToNotifications();
    }

    public void listenToNotifications()
    {
        PGNotificationListener listener = new PGNotificationListener()
                {
                    public void notification(int processId, String channelName, String payload)
                    {
                        System.out.println("Received notification from: "
                                + channelName + ", "
                                + payload);
                        _queueSessionSenderBean.sendMessage(payload);
                    }
                };
            PGDataSource dataSource = new PGDataSource();
            dataSource.setHost("localhost");
            dataSource.setDatabase("db");
            dataSource.setPort(5432);
            dataSource.setUser("user");
            dataSource.setPassword("pass");
        try(PGConnection connection =
                (PGConnection) dataSource.getConnection())
        {
            Statement statement = connection.createStatement();
            statement.execute("LISTEN notifications");
            statement.close();
            connection.addNotificationListener(listener);
            while (true)
            {
                if (Thread.currentThread().isInterrupted())
                {
                    break;
                } 
            }
        }
        catch (Exception e)
        {
            // TODO: handle exception
            e.printStackTrace();
        }
    }   
}

    NotificationStarter.java

    @Singleton
@Startup
public class NotificationsStarter
{
    @EJB
    private NotificationListener _listener;

    @PostConstruct
    public void startListener()
    {
        _listener.start();
    }

    @PreDestroy
    public void shutdown()
    {
        _listener.interrupt();
    }
}

    PushContext.java

    @ApplicationScoped
public class PushContext
{

    @Resource(mappedName = "jms/localNotificationQueue")
    private Queue _localNotificationQueue;

    @Resource(mappedName = "jms/remoteNotificationQueue")
    private Queue _remoteNotificationQueue;

    private Map<String, Set<Session>> _sessions;

    @PostConstruct 
    public void init()
    {
        _sessions = new ConcurrentHashMap<>();
    }

    public void add(Session session, String userUuid)
    {
        _sessions.computeIfAbsent(userUuid, 
                value -> ConcurrentHashMap.newKeySet()).add(session);
    }

    void remove(Session session)
    {
        _sessions.values().forEach(value -> value.removeIf(e -> e.equals(session)));
    }

    public void send(Set<String> userUuids, Message message) throws JMSException 
    {
        String userUuid = message.getStringProperty("userUuid");
        userUuids.add(userUuid);

        Set<Session> userSessions;

        synchronized(_sessions) 
        {
            userSessions = _sessions.entrySet().stream()
                .filter(e -> userUuids.contains(e.getKey()))
                .flatMap(e -> e.getValue().stream())
                .collect(Collectors.toSet());
            for (Session userSession : userSessions) 
            {
                if (userSession.isOpen()) 
                {
                    userSession.getAsyncRemote().sendText(((TextMessage) message).getText());
                }
            }
        }
    }

    public void removeSession(Session session)
    {
        String userUuid = (String)session.getUserProperties().get("userUuid");
        _sessions.remove(userUuid, session);
    }

    public Boolean userHasWebSocketSession(String userUuid)
    {
        Boolean sessionOpen = false;

        Set<String> userUuids = new HashSet<String>();
        userUuids.add(userUuid);

        Set<Session> userSessions;

        synchronized(_sessions) 
        {
            userSessions = _sessions.entrySet().stream()
                .filter(e -> userUuids.contains(e.getKey()))
                .flatMap(e -> e.getValue().stream())
                .collect(Collectors.toSet());
        }
        for (Session userSession : userSessions) 
        {
            if (userSession.isOpen()) 
            {
                sessionOpen = true;
                break;
            }
        }
        return sessionOpen;
    }
}

    QueueSenderSessionBean.java

    @Named
@LocalBean
@Stateless
public class QueueSenderSessionBean
{
    @Resource(mappedName = "jms/localNotificationQueue")
    private Queue _localNotificationQueue;

    @Resource(mappedName = "jms/remoteNotificationQueue")
    private Queue _remoteNotificationQueue;

    @Inject
    @JMSConnectionFactory("jmsConnectionFactory")
    private JMSContext _jmsContext; 

    @Inject
    PushContext _pushContext;

    public void sendMessage(String message) 
    {
        JMSProducer messageProducer =
                _jmsContext.createProducer();
        messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);  
        try
        {        
            String userProperty = "someValue";

            TextMessage textMessage = _jmsContext.createTextMessage(message);
            textMessage.setStringProperty("userProperty", userProperty ); 
            Boolean userIsConnected = 
                    _pushContext.userHasWebSocketSession(userUuid);
            if (!userIsConnected)
            {
                messageProducer.send(_localNotificationQueue, textMessage);
            }
            else
            {
                messageProducer.send(_remoteNotificationQueue, textMessage);
            }
        }
        catch (JMSException e)
        {
            e.printStackTrace();
        }
    }    
}

    NotificationMessageDrivenBean.java 现在只监听远程队列

    @Named
@MessageDriven(mappedName = "jms/remoteNotificationQueue")
public class NotificationMessageDrivenBean implements MessageListener
{
    @Inject
    @NotificationServletJMSMessage
    Event<Message> jmsEvent;

    @Override
    public void onMessage(Message message)
    {
        jmsEvent.fire(message);
    }
}

    New QueueReceiverSessionBean.java 用于接收/使用来自 localNotificationQueue 的消息，并在用户连接到 NotificationEndpoint Web 套接字时将它们放入 remoteNotificationQueue。

    @Named
@LocalBean
@Stateless
public class QueueReceiverSessionBean
{
    @Resource(mappedName = "jms/localNotificationQueue")
    private Queue _localNotificationQueue;

    @Resource(mappedName = "jms/remoteNotificationQueue")
    private Queue _remoteNotificationQueue;

    @Inject
    @JMSConnectionFactory("jmsConnectionFactory")
    private JMSContext _jmsContext; 

    public void receiveQueuedMessages(String userUuidString) throws JMSException
    {
        Set<String> userUuids =
                new HashSet<String>();
        userUuids.add(userUuidString);

        JMSConsumer messageConsumer = 
                _jmsContext.createConsumer(_localNotificationQueue,
                        "userProperty='someValue'",
                        true);

        JMSProducer messageProducer =
                _jmsContext.createProducer();

        Message localMessage =
                messageConsumer.receive(10);
        while(localMessage != null)
        {
            TextMessage textMessage = 
                    _jmsContext.createTextMessage(((TextMessage) localMessage).getText());
            textMessage.setStringProperty("userUuid", userUuidString);            
            messageProducer.send(_remoteNotificationQueue, textMessage);
            localMessage.acknowledge();
            localMessage =
                    messageConsumer.receive(10);
        } 
        messageConsumer.close();
    }

    public void sendMessage(String message) 
    {
        JMSProducer messageProducer =
                _jmsContext.createProducer();
        messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);

        try
        {        
            if (message.startsWith("notification"))
            {
                String messageJson = message.substring(message.indexOf("=") + 1);
                JSONObject notificationJson =
                        new JSONObject(messageJson);
                String userUuid = notificationJson.getString("receivinguseruuid");

                TextMessage textMessage = _jmsContext.createTextMessage(message);
                textMessage.setStringProperty("userUuid", userUuid);            
                messageProducer.send(_remoteNotificationQueue, textMessage);
            }
        }
        catch (JMSException e)
        {
            e.printStackTrace();
        }
    }
}

    NotificationEndpoint.java

    @Named
@ServerEndpoint(value="/notificationEndpoint/{tokenId}")
public class NotificationEndpoint implements Serializable
{

    private static final long serialVersionUID = 1L;
    private static final Set<Session> SESSIONS =
            Collections.synchronizedSet(new HashSet<Session>()); 
    private QueueReceiverSessionBean _senderBean;

    @Inject
    private PushContext _pushContext;

    @Inject
    public NotificationEndpoint(QueueReceiverSessionBean senderBean)
    { 
        _senderBean = senderBean;
    }

    @OnOpen
    public void onOpen(Session session,
            EndpointConfig configurator,
            @PathParam(value = "tokenId") String userUuidString) 
    {
        session.getUserProperties().put("userUuid", userUuidString );        
        _pushContext.add(session, userUuidString);
        try
        {
            _senderBean.receiveQueuedMessages(userUuidString);
        }
        catch (JMSException e)
        {
            e.printStackTrace();
        }
    }


    @OnMessage
    public void onMessage(String message, Session session) 
            throws IOException 
    {
        _senderBean.sendMessage(message);
    }

    @OnClose
    public void onClose(CloseReason reason, Session session) 
    {
        try
        {
            session.close();
        }
        catch (IOException e)
        {
            e.printStackTrace();
        }
        _pushContext.removeSession(session);
    }

    @OnError
    public void error(Session session, Throwable t) 
    {
       t.printStackTrace();
    }

    public static void sendToAllClients(String message) 
    {
        synchronized (SESSIONS) 
        {
            for (Session session : SESSIONS) 
            {
                if (session.isOpen()) 
                {
                    session.getAsyncRemote().sendText(message);
                }
            }
        }
    }

    public void onJMSMessage(@Observes @NotificationServletJMSMessage Message message) 
    {
        Set<String> userUuids = new HashSet<String>();

        try 
        {
            _pushContext.send(userUuids, message);
        } 
        catch (JMSException ex) 
        {
            ex.printStackTrace();
            Logger.getLogger(NotificationEndpoint.class.getName()).
            log(Level.SEVERE, null, ex);
        }     
    }    
}

    注意：此代码在 TomEE 8.0 容器中使用。将 JMSContext 注入 EJB 发现了 TomEE 中的一个错误，即容器无法释放 JMSConnection 资源。问题已添加到TomEE issues tracker

    【讨论】：

      猜你喜欢
      • 2014-09-27
      • 2011-03-27
      • 2012-05-08
      • 2022-12-14
      • 1970-01-01
      • 1970-01-01
      • 2010-12-05
      • 1970-01-01
      • 1970-01-01
      相关资源
      最近更新 更多
      热门标签
      Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode