Wildfly 上的 JMS 2.0 QueueBrowser 不返回消息

Posted 2023-03-10

【中文标题】Wildfly 上的 JMS 2.0 QueueBrowser 不返回消息

【英文标题】：JMS 2.0 QueueBrowser on Wildfly does not return messages

【发布时间】：2015-10-07 05:38:27

【问题描述】：

我有 2 个简单的 EJB bean。 第一个是每秒调用一次的计时器方法。在这种方法中，我将 10 条随机消息添​​加到 TestQueue。

@Singleton
@Startup
public class Timer 

    @Inject
    private JMSContext context;

    @Resource(mappedName = "java:/jms/queue/TestQueue")
    private Queue queue;

    @Schedule(hour = "*", minute = "*", second = "*/1", persistent = false)
    @AccessTimeout(unit = TimeUnit.DAYS, value = 1)
    public void addToQueue() 
        for(int i = 0; i<30; i++)
            context.createProducer().send(queue, "msg " + (new Random().nextInt(100)));
        printQueueSize();
    

    public void printQueueSize() 
        QueueBrowser qb = context.createBrowser(queue);
        Enumeration enumeration = null;
        try 
            enumeration = qb.getEnumeration();
         catch (JMSException e) 
            e.printStackTrace();
        
        int size = 0;
        while(enumeration.hasMoreElements()) 
            enumeration.nextElement();
            size++;
        

        System.out.println("Queue size: " + size);
    

第二个 bean 是 MDB 消费者。我把 Thread.sleep(100) 用来模拟长时间运行的任务，并确保 TestQueue 中有一些消息。

@MessageDriven(
activationConfig = 
        @ActivationConfigProperty(propertyName = "destination", propertyValue = "java:/jms/queue/TestQueue"),
        @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
        @ActivationConfigProperty(propertyName = "maxSession", propertyValue = "1")
) 
public class Consumer implements MessageListener 

    @Override
    public void onMessage(Message msg) 
        try 
            Thread.sleep(100);
         catch (InterruptedException e) 
            e.printStackTrace();
        
    

问题是控制台输出显示：

15:06:29,006 INFO  [stdout] (EJB default - 9) Queue size: 0
15:06:30,006 INFO  [stdout] (EJB default - 10) Queue size: 0
15:06:31,009 INFO  [stdout] (EJB default - 1) Queue size: 0

等等

但在 wildfly 管理控制台中，我可以看到每秒有越来越多的消息：

问题是，为什么 QueueBrowser 返回空 Enumeration？这是 HornetQ 实现中的错误还是有某些原因？

【问题讨论】：

我不确定你的方式是否正确，但我使用JMX方式以编程方式读取队列大小并且它工作正常。

你能提供一些例子吗？

【参考方案1】：

printQueueSize() 返回零，因为您是从 addToQueue() 内部调用它。

您的消息尚未在队列中，因为在进入 addToQueue() 之前启动的事务将在 addToQueue() 调用返回（无错误）之前提交。

JMS 操作是事务性的。

【讨论】：

正如您在屏幕截图中看到的，已经提交了数以百计的事务，JMS 使用者每 100 毫秒（延迟后）调用一次，Wildfly 控制台显示适当的队列大小。交易不是这里的问题。

我同意您应该能够看到以前对 addToQueue 的呼叫的剩余消息。但是，您不会看到当前通话中添加的消息。

请注意，Thread.sleep(100) 意味着您的 MDB 将以几乎每秒 10 条的速度使用消息，这与您添加消息的速度相同。

好点，但是当我将生成的消息数量增加到 20 或 30 时，大小仍然为 0。

Wildfly 10（与 Wildlfy 9 和 HornetQ 相同）和同步 MessageReceiver 现在也有同样的问题，它工作正常，但现在的 ActiveMQ 管理队列不返回排队的消息，例如：Queue managementQueue = ActiveMQJMSClient.createQueue("activemq.management"); 连接到消费者将正确显示排队的消息 count，但不会通过listMessages 返回同一队列的任何消息？！

【参考方案2】：

虽然这不会回答您的问题，但我将展示如何在 HornetQ 中的 EJB 服务中获取队列大小：

    InitialContext initialContext = null;
    try 
        // Step 1. Create an initial context to perform the JNDI lookup.
        initialContext = getContext();

        // Step 2. Perfom a lookup on the queue
        Queue queue = (Queue) initialContext.lookup("jmx/queue/YOUR_QUEUE_NAME");


        // Step 7. Use JMX to retrieve the message counters using the
        // JMSQueueControl
        ObjectName on = ObjectNameBuilder.DEFAULT
                .getJMSQueueObjectName(queue.getQueueName());
        JMXConnector connector = JMXConnectorFactory.connect(new JMXServiceURL("service:jmx:http-remoting-jmx://localhost:9990"), new HashMap<String, Object>());//this is the URL for WildFly 8.2 (should work for all 8.X)
        MBeanServerConnection mbsc = connector.getMBeanServerConnection();
        JMSQueueControl queueControl = MBeanServerInvocationHandler.newProxyInstance(mbsc, on, JMSQueueControl.class, false);

        // Step 8. List the message counters and convert them to
        // MessageCounterInfo data structure.
        String counters = queueControl.listMessageCounter();
        MessageCounterInfo messageCounter = MessageCounterInfo.fromJSON(counters);


        queueControl.getConsumerCount();//this returns consumer count
        queueControl.getMessagesAdded();//returns messages added from the last time we checked
        queueControl.isPaused();//prints out, whether the queue is paused (you can stop a queue from being processed e.g. from JMX)
        queueControl.getMessageCount();//message count so far

     finally 
        // Step 17. Be sure to close our JMS resources!
        if (initialContext != null) 
            initialContext.close();
        
    

【参考方案3】：

我几天来一直在努力解决这个问题，最后发现在我们的例子中，队列被配置为transactional。

这意味着根据 JMS 2 规范，消息在Consumer 端被消费并放入事务中，因此它们不再包含在队列中，也不会出现在其他任何地方，就像一个透明的缓冲区。

见JMS 2 Session API docs:

会话可以被指定为已交易。每个事务会话支持单个系列事务。每个事务*将一组消息发送*和一组消息接收*组合成一个原子工作单元。实际上，事务将会话的*输入消息流*和输出消息流组织成一系列原子单元。当一个事务提交时，它的原子输入单元被确认并且它的相关原子输出单元被发送。如果事务回滚完成，事务发送的消息将被销毁，会话的输入会自动恢复。

事务的输入和输出单元的内容只是*那些在会话的当前事务中产生和消费的消息。

只需在 Wildfly 的配置（通常为 standalone-full.xml）中将属性 transaction 设置为 none 即可解决问题：

pooled-connection-factory name="activemq-ra" entries="java:/JmsXA java:jboss/DefaultJMSConnectionFactory" connectors="in-vm" transaction="none"

【讨论】：

另请参阅***.com/a/25246084/160799 以获得有关 HornetQ 的实用解释（与 Artemis/ActiveMQ 相同）