无法从 jms activemq 队列中删除消息

Question

我是 jms 技术的新手。我正在使用activeMQ 控制台来监控队列。我可以创建包含消息的队列。但是当我尝试删除某些队列时..一个例外来了。尝试了很多东西，但一切都在脉络..下面是我的代码：

代码

   BrokerService brokerService = new BrokerService();
        try {
            brokerService.start(true);
            brokerService.start();
            brokerService.autoStart();
            brokerService.setUseJmx(true);
            brokerService.addConnector("tcp://localhost:61616");
            Thread.sleep(10000);
            System.out.println("brokerService : "+brokerService.isStarted());
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }


        BrokerFacadeSupport facade=   new LocalBrokerFacade(brokerService);
        try {
            QueueViewMBean queue = facade.getQueue(queue_name);
            if(queue==null) {
                System.out.println("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!");
            }
                    int count = this.jmsTemplate.browse(queue_name, new BrowserCallback<Integer>() {
                public Integer doInJms(final Session session, final QueueBrowser browser) throws JMSException {
                    Queue queue1 = browser.getQueue();
                    Enumeration enumeration = browser.getEnumeration();
                    int counter = 0;
                    while (enumeration.hasMoreElements()) {
                        Message msg = (Message) enumeration.nextElement();
                        msg.acknowledge();
                        ActiveMQTextMessage atm  = (ActiveMQTextMessage) msg;
                        atm.setDroppable(true);
                        atm.setReadOnlyProperties(false);
                        atm.setReadOnlyBody(false);
                        atm.acknowledge();
                        msg.setBooleanProperty("readOnlyProperties", false);
                                    msg.setBooleanProperty("readOnlyBody", false);
                                    msg.setBooleanProperty("droppable", true);
                        Enumeration enum_ = atm.getPropertyNames();
                        while(enum_.hasMoreElements()) {
                            String name  = (String) enum_.nextElement();
                            System.out.println("## : "+name);
                        }
                        try {
                            System.out.println("--"+atm.getJMSMessageID());
                            System.out.println();
                            queue.removeMessage(atm.getJMSMessageID());
                        }catch(Exception ex) {
                            ex.printStackTrace();
                        }
                        System.out.println(String.format("\tFound : %s", msg));
                        counter += 1;
                    }
                    return counter;
                }
            });
            queue.purge();
        }catch(Exception ex) {
            ex.printStackTrace();
        }

输出

输出到System.out.println(String.format("\tFound : %s", msg)); 是：

Found : ActiveMQTextMessage {commandId = 5, responseRequired = true, messageId = ID:CRM-PC-50101-1528866712471-1:1:1:1:1, originalDestination = null, originalTransactionId = null, producerId = ID:CRM-PC-50101-1528866712471-1:1:1:1, destination = queue://testNexus, transactionId = null, expiration = 0, timestamp = 1528866713408, arrival = 0, brokerInTime = 1528866713410, brokerOutTime = 1529047482640, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = org.apache.activemq.util.ByteSequence@12b82970, marshalledProperties = org.apache.activemq.util.ByteSequence@49abe550, dataStructure = null, redeliveryCounter = 0, size = 0, properties = {_type=com.crm.jms.SampleObject}, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false, text = {"msg":"hello world"}}

最终输出控制台

==>   _type : com.crm.jms.SampleObject
--ID:Crm-PC-50101-1528866712471-1:1:1:1:1
java.lang.NullPointerException
    at com.crm.jms.controller.QueueController$1.doInJms(QueueController.java:171)
    at com.crm.jms.controller.QueueController$1.doInJms(QueueController.java:1)
    at org.springframework.jms.core.JmsTemplate$14.doInJms(JmsTemplate.java:1033)
    at org.springframework.jms.core.JmsTemplate.execute(JmsTemplate.java:484)
    at org.springframework.jms.core.JmsTemplate.browseSelected(JmsTemplate.java:1027)
    at org.springframework.jms.core.JmsTemplate.browse(JmsTemplate.java:989)
    at com.crm.jms.controller.QueueController.deleteQueue(QueueController.java:153)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
.
.
.
.
    Found : ActiveMQTextMessage {commandId = 5, responseRequired = true, messageId = ID:Crm-PC-50101-1528866712471-1:1:1:1:1, originalDestination = null, originalTransactionId = null, producerId = ID:Crm-PC-50101-1528866712471-1:1:1:1, destination = queue://testNexus, transactionId = null, expiration = 0, timestamp = 1528866713408, arrival = 0, brokerInTime = 1528866713410, brokerOutTime = 1529049245130, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = org.apache.activemq.util.ByteSequence@5b55f70f, marshalledProperties = org.apache.activemq.util.ByteSequence@3e116ae6, dataStructure = null, redeliveryCounter = 0, size = 0, properties = {_type=com.crm.jms.SampleObject}, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false, text = {"msg":"hello world"}}
java.lang.NullPointerException
    at com.crm.jms.controller.QueueController.deleteQueue(QueueController.java:181)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:205)
    at org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:133)
    at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:97)
    at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:827)
    at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:738)
    at org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:85)
.
.
.
.

Answer 1

您为什么要确认您正在浏览（而不是消费）的消息？

Enumeration enumeration = browser.getEnumeration();
...
Message msg = (Message) enumeration.nextElement();
msg.acknowledge();

正如您所注意到的，在 javax.jms.Queue 上确实没有 delete()/remove() 方法，但是您可以通过使用所需的消息（仍然使用纯 JMS API）来达到相同的结果：

MessageConsumer consumer = session.createConsumer(myQueue, "JMSMessageID = '" + messageId + "'");
Message message = consumer.receive(TIMEOUT);
message.acknowledge();

Answer 2

问题在于您没有启动BrokerService，因此facade.getQueue(queueName) 方法将返回null。为什么？因为外观对象不知道您要的是哪个活动的 mq 服务器。

当然，因为您正在使用jmsTemplate 连接到正在运行的代理服务器，所以您会获得一些关于您的消息的信息。

在queue.purge(); 和queue.removeMessage(msg.getJMSMessageID()); 上抛出异常，因为queue 是null。

我建议删除BrokerService 和BrokerServiceFacade 的东西，并通过browser.getQueue() 方法从QueueBrowser 参数中获取回调中的Queue 对象。