带有 XML 配置的 ActiveMQ 嵌入式代理主题到队列桥

Question

我的目标 + 进度

我对使用 ActiveMQ 将消息发布到主题并将其桥接到多个队列感兴趣。通过提供一个包含复合主题的 xml-config，我已经设法通过命令行代理实现了这一点：

<destinationInterceptors>
    <virtualDestinationInterceptor>
        <virtualDestinations>
            <compositeTopic name="LOCAL.EC.T">
                <forwardTo>
                    <queue physicalName="LOCAL.EC.Q.1" />
                    <queue physicalName="LOCAL.EC.Q.2" />                       
                </forwardTo>
            </compositeTopic>
        </virtualDestinations>
    </virtualDestinationInterceptor>
</destinationInterceptors>

使用这个启动命令：activemq start xbean:amq.xml。在本例中，amq.xml 是我的活动 MQ XML 配置文件。这里有更多关于 xml 配置的内容：http://activemq.apache.org/xml-configuration.html.

我的问题

如果我使用 Web 控制台向上述主题发布消息，它会按预期显示在两个队列中。但现在我想改用嵌入式代理。

---

在下面的代码中，我可以看出它正在使用我的 amq.xml 文件（当我更改它时，我得到了相关的错误），但是当我发布到主题时，队列中的接收将永远阻塞.如果我发布和接收相同的主题或队列，一切正常。 为什么我的复合主题在这里不起作用？

---

//Create the broker using the xbean configuration and start it.
brokerService = BrokerFactory.createBroker(new URI("xbean:amq.xml"));
brokerService.setBrokerName("localhost");
brokerService.setPersistent(false);
brokerService.setDeleteAllMessagesOnStartup(true);
brokerService.setUseJmx(false);
brokerService.start();

//Create the connection factory and JMS template.
connectionFactory = new ActiveMQConnectionFactory();
connectionFactory.setBrokerURL("vm://localhost?create=false&jms.redeliveryPolicy.maximumRedeliveries=-1");

//Send the message to the topic.
template = new JmsTemplate(connectionFactory);
ActiveMQMapMessage message = new ActiveMQMapMessage();
message.setString("batch", "Hello World!");
template.convertAndSend("LOCAL.EC.T",message);

//Read the message from the queues.
final Message receive = template.receive("LOCAL.EC.Q.1");
MapMessage received = (MapMessage)receive;
System.out.println(received.getString("batch"));  

Answer 1

我在一个网站上找到了一个示例，该示例演示了如何以稍微不同的方式生成消息。不幸的是，我关闭了标签，找不到引用它的链接。

无论哪种方式，我都将这种消息生产风格与我的 xbean 配置相结合，现在一切正常。这是工作代码：

brokerService = BrokerFactory.createBroker(new URI("xbean:amq.xml"));
brokerService.setPersistent(false);
brokerService.setDeleteAllMessagesOnStartup(true);
brokerService.setUseJmx(false);
brokerService.start();

connectionFactory = new ActiveMQConnectionFactory("vm://localhost");
template = new JmsTemplate(connectionFactory);

//Create a connection.
Connection connection = connectionFactory.createConnection();
connection.start();

//Create a session.
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

//Create a topic and publish to it - it should be linked to 4 queues by the configuration.
System.out.println("Posting message to topic:");
Destination destination = session.createTopic("LOCAL.EC.T");
MessageProducer producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
ActiveMQMapMessage message = new ActiveMQMapMessage();
message.setString("batch", "Hello World!");
producer.send(message);

//Read the message from the queues.
System.out.println("Q1: " + ((MapMessage)receive("LOCAL.EC.Q.1")).getString("batch"));
System.out.println("Q2: " + ((MapMessage)receive("LOCAL.EC.Q.2")).getString("batch"));
System.out.println("Q3: " + ((MapMessage)receive("LOCAL.EC.Q.3")).getString("batch"));
System.out.println("Q4: " + ((MapMessage)receive("LOCAL.EC.Q.4")).getString("batch"));

输出：

向主题发布消息：

Q1：世界你好！

Q2：世界你好！

Q3：世界你好！

Q4：世界你好！

结束申请。

希望它可以帮助其他人！