ActiveMQ使用/转发来自另一个ActiveMQ实例的消息

Question

我有两个代理A和B。如果我想将消息从A转发到B，一切都很简单。我只需要网络连接器在一个代理配置如下：

<networkConnectors>
<networkConnector staticBridge="true" userName="user" password="pass" uri="static://(tcp://B:61616)">
<staticallyIncludedDestinations>
<queue physicalName="QUEUE.TO.FORWARD.MESSAGE" />
</staticallyIncludedDestinations>
</networkConnector>
</networkConnectors>

如果我想从其他队列(让我们将其命名为queue.to.consume(中消费来自broker B的消息，我会很强硬。我只需要做同样的事情，但双工设置为true，只需在queue.to.consume上侦听即可。在broker A上消费，如下所示：

<networkConnectors>
<networkConnector name="from-B-to-A" staticBridge="true" duplex="true" userName="user" password="pass" uri="static://(tcp://B:61616)">
<staticallyIncludedDestinations>
<queue physicalName="QUEUE.TO.CONSUME" />
</staticallyIncludedDestinations>
</networkConnector>
<networkConnector staticBridge="true" userName="user" password="pass" uri="static://(tcp://B:61616)">
<staticallyIncludedDestinations>
<queue physicalName="QUEUE.TO.FORWARD.MESSAGE" />
</staticallyIncludedDestinations>
</networkConnector>
</networkConnectors>

但它并没有像我预期的那样起作用。似乎只有每一秒的信息都被忽略了，剩下的都丢失了。令人惊讶的是，这在broker B上创建了两个消费者QUEUE.TO.CONUSE和我假设其中一个消费者在不转发到broker A的情况下消费消息。如何在broker A上创建桥接器，允许我消费来自broker B的消息而不丢失消息。目前还不能在代理程序B中创建网络连接器。我还尝试过创建这样的入站队列桥：

<jmsBridgeConnectors>
<jmsQueueConnector outboundQueueConnectionFactory="#remoteBroker" localUsername="user" localPassword="password">
<inboundQueueBridges>
<inboundQueueBridge inboundQueueName="QUEUE.TO.CONSUME" localQueueName="QUEUE.TO.CONSUME" />
</inboundQueueBridges>
</jmsQueueConnector>
</jmsBridgeConnectors>
...
</broker>
<bean id="remoteBroker" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="failover://(nio:B:61616)" />
<property name="userName" value="user" />
<property name="password" value="password" />
</bean>

此配置在远程代理B上创建消费者，但它不使用任何挂起为已排队的消息，并且不会发生任何事情。Broker A仍然没有接收到任何到其本地队列的消息。

Answer 1

好的，我想明白了。我刚刚使用嵌入式ApacheCamel来定义到远程主机的路由，它看起来像这样(conf目录中的Camel.xml(：

<beans
xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:camel="http://camel.apache.org/schema/spring"
xsi:schemaLocation="
http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<camelContext id="context" xmlns="http://camel.apache.org/schema/spring">
<route>
<from uri="remoteBroker:queue:QUEUE.TO.CONSUME"/>
<to uri="localBroker:queue:QUEUE.TO.CONSUME"/>
</route>
</camelContext>
<bean id="remoteBroker" class="org.apache.activemq.camel.component.ActiveMQComponent">
<property name="connectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://B:61616"/>
<property name="userName" value="user"/>
<property name="password" value="password"/>
</bean>
</property>
</bean>
<bean id="localBroker" class="org.apache.activemq.camel.component.ActiveMQComponent">
<property name="connectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost"/>
</bean>
</property>
</bean>
</beans>

其中localhost i broker A.在activemq.xml中：

<import resource="camel.xml"/>