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>

如果我想从其他队列中使用来自代理 B 的消息，我会犹豫不决（让我们将其命名为 QUEUE。自。消费）我只需要做同样的事情，但双工设置为true，只需听队列。自。在代理 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>

但是它不像我预期的那样工作。似乎每一秒钟只有一条信息被转发，其余的都丢失了。令人惊讶的是，这在代理B队列上创建了两个消费者。如何在代理A上创建桥，允许我在不丢失消息情况下使用来自代理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 上创建使用者，但它不会消耗任何消息，这些消息只是挂起为排队状态，没有任何反应。代理 A 仍不会接收到其本地队列的任何消息。

Answer 1

好吧，我明白了。我刚刚使用嵌入式Apache Camel来定义到远程主机的路由，它看起来像这样(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"/>