ActiveMQ JMSXGroupId属性无效

Question

我想通过使用JMSXGroupId属性对消息进行分组。但是消费者无需分组即可获取所有队列消息

我有一个本地ActiveMQ实例（版本5.15.9，Windows 10，Java 8）。我已经编写了DoSend Java类来发送文本消息，并编写了DoRead Java类来检索文本消息

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class DoSend {
    public static void main(String[] args) throws JMSException {
        ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");

        Connection connection = null;
        try {
            connection = factory.createConnection("admin", "admin");
            connection.start();

            Session session = null;
            try {
                session = connection.createSession(true, Session.SESSION_TRANSACTED);

                MessageProducer producer = null;
                try {
                    Destination destination = session.createQueue("TEST");
                    producer = session.createProducer(destination);

                    int index = 0;
                    while (true) {
                        index += 1;

                        TextMessage message = session.createTextMessage(String.valueOf(index));

                        message.setStringProperty("JMSXGroupID", String.valueOf(index % 3));
                        producer.send(destination, message);
                        session.commit();

                        Thread.sleep(1000);
                    }
                } finally {
                    if (producer != null) {
                        producer.close();
                    }
                }

            } catch (Throwable th) {
                th.printStackTrace();

                if (session != null) {
                    session.rollback();
                }

                throw new RuntimeException(th);
            } finally {
                if (session != null) {
                    session.close();
                }
            }
        } finally {
            if (connection != null) {
                connection.close();
            }
        }
    }
}

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class DoRead {
    public static void main(String[] args) throws JMSException {
        final ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");

        System.out.println("START " + Thread.currentThread());
        try {
            Connection connection = null;
            try {
                connection = factory.createConnection("admin", "admin");
                connection.start();

                Session session = null;
                try {
                    session = connection.createSession(true, Session.SESSION_TRANSACTED);

                    Destination destination = session.createQueue("TEST");
                    MessageConsumer consumer = null;
                    try {
                        consumer = session.createConsumer(destination);

                        TextMessage message;
                        while ((message = (TextMessage) consumer.receive(10000)) != null) {
                            System.out.println(Thread.currentThread() + ": message.index = " + message.getText());
                            System.out.println(Thread.currentThread() + ": message.JMSXGroupID = " + message.getStringProperty("JMSXGroupID"));
                            session.commit();
                        }
                    } finally {
                        if (consumer != null) {
                            consumer.close();
                        }
                    }
                } finally {
                    if (session != null) {
                        session.close();
                    }
                }
            } finally {
                if (connection != null) {
                    connection.close();
                }
            }
        } catch (JMSException e) {
            e.printStackTrace();
        } finally {
            System.out.println("FINISH " + Thread.currentThread());
        }
    }
}

[我希望任何使用者都使用相同的JMSXGroupId检索消息，但是现在第一个使用者检索所有消息，而其他使用者什么也不检索。

Answer 1

此标题的目的是将单个接收器与组相连。因此，只有一个接收者将获得所有具有相同组ID的消息。通常，您还希望添加JMSXGroupSeq标头以指示组中的消息ID。此标头的通用目的是要在一个节点（单个接收器）上执行一些聚合逻辑。

请参见ActiveMq page了解详情或JBOSS page。