java 如何在activemq中杀死消费者

How to kill consumers in activemq

提问by Chris Ridmann

I am trying to get rid of all of the "Number of Consumers" in a certain queue. Whenever I purge/delete the queue, the number of consumers still remain if I ever create that queue with the same name again. Even with 0 pending messages, there are still 6 consumers.

我试图摆脱某个队列中的所有“消费者数量”。每当我清除/删除队列时，如果我再次创建具有相同名称的队列，消费者的数量仍然存在。即使有 0 个待处理消息，仍然有 6 个消费者。

My problem may have stemmed in my java code while not closing the session or connection.

我的问题可能出在我的 java 代码中，而没有关闭会话或连接。

I have tried both restarting and reinstalling the server.

我已经尝试重新启动和重新安装服务器。

Here is my producer code:

这是我的生产者代码：

 private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;

    public static String addElementToQueue(String queueName,String param1, String param2) throws JMSException, NamingException {
  // Getting JMS connection from the server and starting it
        ConnectionFactory connectionFactory =
                new ActiveMQConnectionFactory(url);
        Connection connection = connectionFactory.createConnection();

// JMS messages are sent and received using a Session. We will
        // create here a non-transactional session object. If you want
        // to use transactions you should set the first parameter to 'true'
        Session session = connection.createSession(false,
                Session.AUTO_ACKNOWLEDGE);

        // Destination represents here our queue on the
        // JMS server. You don't have to do anything special on the
        // server to create it, it will be created automatically.
        Destination destination = session.createQueue(queueName);

        // MessageProducer is used for sending messages (as opposed
        // to MessageConsumer which is used for receiving them)
        MessageProducer producer = session.createProducer(destination);   

        String queueMessage = param1+ "-" + param2;

        TextMessage message = session.createTextMessage(queueMessage);

        // Here we are sending the message!
        producer.send(message);

        connection.close();
        session.close();      // added after problem came up
        producer.close();     // added after problem came up

        return commandID;
}

Here is my consumer code:

这是我的消费者代码：

 // URL of the JMS server
    private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;

    public static Pair consumeNextElement(String queueName) throws JMSException {
        // Getting JMS connection from the server
        ConnectionFactory connectionFactory
                = new ActiveMQConnectionFactory(url);
        Connection connection = connectionFactory.createConnection();
        connection.start();   

        // Creating session for seding messages
        Session session = connection.createSession(false,
                Session.AUTO_ACKNOWLEDGE);

        // Getting the queue
        Destination destination = session.createQueue(queueName);

        // MessageConsumer is used for receiving (consuming) messages
        MessageConsumer consumer = session.createConsumer(destination);


        // Here we receive the message.
        // By default this call is blocking, which means it will wait
        // for a message to arrive on the queue.
        Message message = consumer.receive();

        // There are many types of Message and TextMessage
        // is just one of them. Producer sent us a TextMessage
        // so we must cast to it to get access to its .getText()
        // method.

        String[] parts = ((TextMessage)message).getText().split("-");
        Pair retVal = new Pair(parts[0], parts[1]);

        connection.close();
        session.close();        // added after problem came up
        consumer.close();      // added after problem came up  

        return retVal;
    }

Any thoughts?

有什么想法吗？

Thanks.

谢谢。

采纳答案by Brian Henry

The number of consumers is the number of listeners on the queue. Purging the queue should only remove the enqueued messages - those consumers listening will be unaffected.

消费者的数量是队列中侦听器的数量。清除队列应该只删除排队的消息 - 那些收听的消费者不会受到影响。

The ability of the consumer to maintain/re-establish a connection may depend on the transport used to connect, and settings for the transportmay allow for some tweaking of connection properties.

消费者维护/重新建立连接的能力可能取决于用于连接的传输，并且传输的设置可能允许对连接属性进行一些调整。

I frankly don't have much experience with these, but you might investigate Advisory Messagesas a means to help debug your connections. The JMX interface or web console don't appear to be helpful beyond reporting consumer counts.

坦率地说，我在这些方面没有太多经验，但您可以调查咨询消息作为帮助调试连接的一种方式。除了报告消费者数量之外，JMX 界面或 Web 控制台似乎没有帮助。