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

Question

我是 jms 技术的新手。我正在使用 activeMQ 控制台来监控队列。我能够创建其中包含消息的队列。但是当我尝试删除某些队列时……出现异常。尝试了很多东西，但都在 vein.. 下面是我的代码:

代码

   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();