Java 如何让 MessageListener 在收到某个消息时停止监听 JMS 中的消息？

How to make MessageListener stop listening for messages in JMS when it receives a certain message?

提问by theGuardian

I have a message listener that is receiving some TextMessages. When it receives an ObjectMessage, I want it to stop listening to the queue. My problem is that when I call, consumer.close() inside the onMessage(Message msg) method, the ObjectMessage does not seem to be removed from the Queue. If I use some marker to tell the consuemr to close after the onMessage() method, the listener may consume another message before it actually closes. Any suggestions? Here is some code. The Session, Connection, and InitialContext have not been closed yet.

我有一个正在接收一些 TextMessage 的消息侦听器。当它收到 ObjectMessage 时，我希望它停止侦听队列。我的问题是，当我在 onMessage(Message msg) 方法中调用 consumer.close() 时，ObjectMessage 似乎没有从队列中删除。如果我在 onMessage() 方法之后使用一些标记告诉消费者关闭，则侦听器可能会在实际关闭之前消耗另一条消息。有什么建议？这是一些代码。Session、Connection 和 InitialContext 尚未关闭。

public class MyListener implements MessageListener{
    MessageConsumer consumer;

    public MyListener(MessageConsumer mc){
        consumer = mc;
    }

    @Override
    public void onMessage(Message msg) {
        try{
            if(msg instanceof ObjectMessage){
                consumer.close();
            }
            if (msg instanceof TextMessage){
                TextMessage tmsg = (TextMessage) msg;
                String xml = tmsg.getText();
                // do some stuff                
            }

       }catch(Exception e){
           e.printStackTrace();
       }
    }

采纳答案by Beryllium

Do not use an asynchronous MessageListener.

不要使用异步MessageListener.

Instead use the normal synchronous receivemethod in your main thread in a loop. If you get your special message, you can acknowledge and break from the loop to close the session, and to terminate the program.

而是receive在循环中的主线程中使用普通的同步方法。如果您收到您的特殊消息，您可以确认并中断循环以关闭会话并终止程序。

回答by aUserHimself

Read the documentationfirst. You may have another thread(s) accessing MessageConsumerand the thread that calls close()will block until the completion of other(s).

首先阅读文档。您可能有另一个线程正在访问，MessageConsumer并且调用的线程close()将阻塞，直到其他线程完成。

回答by Miguel Abraham

After a few hours trying to solve this, I think I found a way to stop an asynchronous message consumer (MessageListener). The solution involves using Java locks (synchronized statements, and wait/notify methods).

在尝试解决这个问题几个小时后，我想我找到了一种方法来停止异步消息使用者 (MessageListener)。该解决方案涉及使用 Java 锁（同步语句和等待/通知方法）。

First, on your main thread you need to lock the message listener after starting your JMS connection and invoke the message listener “wait” method. On your message listener, you need to lock again the message listener and then invoke the “notify all” method.

首先，在您的主线程上，您需要在启动 JMS 连接后锁定消息侦听器并调用消息侦听器“wait”方法。在您的消息侦听器上，您需要再次锁定消息侦听器，然后调用“notify all”方法。

  // Main thread ...
  public static void main(String[] args) {
    // ...
    try {
      Connection jmsConn;
      MessageConsumer msgConsumer;
      MessageListener msgListener;
      // ...
      msgConsumer.setMessageListener(msgListener);
      // ...
      synchronized (msgListener) {
        jmsConn.start();
        msgListener.wait();
      }
      jmsConn.stop();
      //...
    } catch (Exception e) {
      // ...
    }
  }


  // MessageListener onMessage...
  public void onMessage(Message jmsMsg) {
    try {
      // ...
      synchronized (this) {
        this.notifyAll();
      }
    } catch (Exception e) {
      // ...
    }
  }

Miguel Abraham

米格尔亚伯拉罕

回答by jason.kaisersmith

This may be a little old, but as I had the same issue and came across it then I thought I would post my findings to help others.

这可能有点旧，但由于我遇到了同样的问题并遇到了它，所以我想我会发布我的发现来帮助其他人。

I had the same issues as the question, I had created a JMS receiver class which set an async listener;

我遇到了与问题相同的问题，我创建了一个设置异步侦听器的 JMS 接收器类；

TopicSubscriber receiver = myTopicSession.createSubscriber(myTopic);  
JmsMessageListener listener = new JmsMessageListener();
receiver.setMessageListener(listener);

And then I could not terminate the listener in a nice way.

然后我无法以一种很好的方式终止侦听器。

I found the solution was to actually close the connection to my topic. And that this would terminate the listener thread as well.

我发现解决方案是实际关闭与我的主题的连接。并且这也会终止侦听器线程。

myTopicConnection.close();

It meant that in my main thread I had to keep a link to the JMS receiver class I had created and then call a close() method to shut it down.

这意味着在我的主线程中，我必须保持到我创建的 JMS 接收器类的链接，然后调用 close() 方法将其关闭。

回答by yaroslavpalamar

It is correct that you can't call connection.stop()or connection.close()from onMessage()according to JMS specification, but you can call connection.close()and connection.stop()from other thread, so in my case I just setup volatile variable from onMessage()when need to stop the connection, and check this variable in other thread where I can call connection.stop()and connection.close()without getting the exception or deadlock.

这是正确的，你不能打电话connection.stop()或connection.close()从onMessage()根据JMS规范，但你可以调用connection.close()和connection.stop()来自其他线程在我的情况，所以从我刚刚安装volatile变量onMessage()时，需要停止连接，并在另一个线程，其中检查该变量我可以调用connection.stop()并且connection.close()不会出现异常或死锁。

> From JMS 2.0 spec:

> 从 JMS 2.0 规范：

6.1.5. Pausing delivery of incoming messages If any message listeners are running when stop is invoked, stop must wait until all of them have returned before it may return. While these message listeners are completing, they must have the full services of the connection available to them. A message listener must not attempt to stop its own connection as this would lead to deadlock. The JMS provider must detect this and throw a javax.jms.IllegalStateException.

> From JMS 1.1:

4.3.4 Pausing Delivery of Incoming Messages If MessageListeners are running when stop is invoked, stop must wait until all of them have returned before it may return. While these MessageListeners are completing, they must have the full services of the connection available to them.

6.1.5. 暂停传入消息的传递 如果调用 stop 时有任何消息侦听器正在运行，则 stop 必须等到所有消息侦听器都返回后才能返回。当这些消息侦听器完成时，它们必须具有可用的连接的完整服务。消息侦听器不得尝试停止自己的连接，因为这会导致死锁。JMS 提供者必须检测到这一点并抛出 javax.jms.IllegalStateException。

> 从 JMS 1.1：

4.3.4 暂停传入消息的传递 如果在调用 stop 时 MessageListener 正在运行，则 stop 必须等到所有消息都返回后才能返回。当这些 MessageListener 完成时，它们必须具有可用的连接的完整服务。