代理网络

功能 > 集群 > 代理网络

为了提供大型消息结构的巨大可扩展性，您通常希望允许许多代理连接到一个网络中，这样您就可以拥有尽可能多的客户端，所有这些客户端在逻辑上都连接在一起——并且根据您的客户端数量和网络拓扑运行尽可能多的消息代理。

如果您使用的是客户端/服务器或中心/辐射式拓扑，那么您连接的代理将成为单点故障，这也是需要网络（或代理集群）的另一个原因，这样您就可以在任何特定代理、机器或子网发生故障的情况下存活下来。

从 ActiveMQ Classic 1.1 开始，支持代理网络，这使我们能够在代理网络中支持分布式队列和主题。

这使客户端能够连接到网络中的任何代理——如果出现故障，则可以故障转移到另一个代理——从客户端的角度来看，提供了一个HA代理集群。

注意：默认情况下，网络连接是单向的——建立连接的代理将消息传递到与其连接的代理。从 ActiveMQ Classic 5.x 版本开始，可以可选地启用网络连接以实现双向，这对于中心和辐射式架构很有用，在这种架构中，中心位于防火墙后面等。

配置代理网络

配置代理网络的最简单方法是通过XML 配置。创建代理网络主要有两种方法：

• 使用硬编码的 networkConnector 元素列表。

• 使用发现来检测代理（组播或集合点）。

使用固定 URI 列表的示例

以下是用固定 URI 列表的示例

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://activemq.org/config/1.0">

  <broker brokerName="receiver" persistent="false" useJmx="false">  
    <networkConnectors>
      <networkConnector uri="static:(tcp://127.0.0.1:62001)"/>
    </networkConnectors>

    <persistenceAdapter>
      <memoryPersistenceAdapter/>
    </persistenceAdapter>

   <transportConnectors>
      <transportConnector uri="tcp://127.0.0.1:62002"/>
    </transportConnectors>
  </broker>

</beans>

ActiveMQ Classic 还支持除 TCP 以外的其他传输用于网络连接器，例如 HTTP。

使用组播发现的示例

此示例使用组播发现

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://activemq.org/config/1.0">

  <broker name="sender" persistent="false" useJmx="false">  
    <networkConnectors>
      <networkConnector uri="multicast://default"/>
    </networkConnectors>

    <persistenceAdapter>
      <memoryPersistenceAdapter/>
    </persistenceAdapter>

  <transportConnectors>
      <transportConnector uri="tcp://127.0.0.1:0" discoveryUri="multicast://default"/>
    </transportConnectors>
  </broker>

</beans>

启动网络连接器

默认情况下，网络连接器在代理启动顺序的一部分中串行启动。当某些网络缓慢时，它们会阻止其他网络及时启动。版本 5.5 支持代理属性 networkConnectorStartAsync=”true”，这将导致代理使用执行程序并行启动网络连接器，异步于代理启动。

静态发现

使用static:发现，您可以硬编码代理 URL 列表。将为每个 URL 创建一个网络连接器。

<networkConnectors>
  <networkConnector uri="static:(tcp://host1:61616,tcp://host2:61616,tcp://..)"/>
</networkConnectors>

您可以在静态网络连接器上设置一些有用的属性以进行重试

例如：

uri="static:(tcp://host1:61616,tcp://host2:61616)?maxReconnectDelay=5000&useExponentialBackOff=false"

主从发现

代理网络的常见配置选项是在代理和 n+1 代理对（主/从）之间建立网络桥。典型的配置涉及使用failover:传输，但有一些其他非直观的选项必须配置才能按预期工作。为此，ActiveMQ Classic v5.6+ 提供了一个便捷的发现代理，可以使用masterslave:传输前缀指定。

<networkConnectors>
  <networkConnector uri="masterslave:(tcp://host1:61616,tcp://host2:61616,tcp://..)"/>
</networkConnectors>

URI 按以下顺序列出：MASTER、SLAVE1、SLAVE2…SLAVE

与static:相同的配置选项也适用于masterslave:

NetworkConnector 属性

可靠性

代理网络对消息进行可靠的存储转发。如果源是持久的，则队列上的持久消息或持久主题订阅将保留持久性保证。
但是，网络无法在源不是持久的情况下添加持久性。非持久主题订阅和临时目标（队列和主题）在定义上都是非持久的。当非持久
源联网时，如果发生故障，则可能会丢失正在传输的消息。

排序

代理网络不会保留消息的总排序。总排序适用于单个消费者，但网络桥梁引入了第二个消费者。此外，网络桥梁消费者通过 producer.send(..) 转发消息，因此它们从转发代理队列的头部转到目标队列的尾部。如果单个消费者在联网代理之间移动，如果所有消息始终跟随消费者，则可能会保留总顺序，但这在消息积压量很大的情况下很难保证。

何时使用和不使用管道订阅

ActiveMQ Classic 依赖于有关活动消费者（订阅）的信息将消息传递到网络中。代理对来自远程（联网）代理的订阅的解释方式与对来自本地客户端连接的订阅的解释方式相同，并将任何相关消息的副本路由到每个订阅。使用主题订阅，并且有多个远程订阅时，远程代理会将每条消息副本解释为有效，因此，当它依次将消息路由到自己的本地连接时，会出现重复。因此，默认的管道行为将所有匹配的订阅信息合并起来，以防止重复项在网络中流动。使用此默认行为，远程代理上的 N 个订阅对联网代理看起来像单个订阅。

但是，如果只使用队列，重复订阅是一个有用的功能。因为负载均衡算法会尝试平均分配消息负载，所以只有在设置标志 conduitSubscriptions=false 的情况下，跨网络的消费者才能平等地共享消息负载。以下是一个例子。假设您有两个代理，A 和 B，它们通过转发桥连接在一起。连接到代理 A，您有一个消费者订阅了一个名为 Q.TEST 的队列。连接到代理 B，您有两个消费者也订阅了 Q.TEST。所有消费者都具有相同的优先级。然后，您在代理 A 上启动一个生产者，它将 30 条消息写入 Q.TEST。默认情况下，（conduitSubscriptions=true），15 条消息将发送到代理 A 上的消费者，而剩余的 15 条消息将发送到代理 B 上的两个消费者。消息负载没有在所有三个消费者之间平均分配，因为默认情况下，代理 A 将代理 B 上的两个订阅视为一个。如果您将 conduitSubscriptions 设置为 false，那么三个消费者中的每一个都将收到 10 条消息。

双工网络连接器

默认情况下，网络桥按需在一个方向上通过单个连接转发消息。当 duplex=true 时，同一个连接用于网络桥在相反方向上的传输，从而形成一个双向桥。网络桥配置被传播到另一个代理，因此双工桥是原始桥的精确副本。

假设有两个代理，代理 A 和代理 B，在 A 到 B 上的双工桥与在 A 到 B 上的默认桥以及在 B 到 A 上的默认桥相同。

注意，如果您想在两个代理之间配置多个双工网络桥，以提高吞吐量或对主题和队列进行分区，则必须为每个桥提供唯一的名称。

<networkConnectors>
        <networkConnector name="SYSTEM1" duplex="true" uri="static:(tcp://10.x.x.x:61616)">
                <dynamicallyIncludedDestinations>
                        <topic physicalName="outgoing.System1" />
                </dynamicallyIncludedDestinations>
        </networkConnector>
        <networkConnector name="SYSTEM2" duplex="true" uri="static:(tcp://10.x.x.x:61616)">
                <dynamicallyIncludedDestinations>
                        <topic physicalName="outgoing.System2"/>
                </dynamicallyIncludedDestinations>
        </networkConnector>
  </networkConnectors>

管道订阅和消费者选择器

管道订阅会忽略本地代理上的消费者选择器，并将所有消息发送到远程代理。然后，在消息被分派到消费者之前，选择器会在远程代理上解析。这个概念可能会在使用选择器的多代理网络中使用队列进行消费时造成一些问题。想象一下，当您有一个生产者代理将消息转发到两个接收代理，并且这两个代理中的每一个都具有一个具有不同选择器的消费者时的情况。由于生产者代理侧没有评估任何选择器，因此您最终可能会将所有消息发送到其中一个代理，因此具有某些属性的消息将不会被消费。如果您需要支持这种用例，请关闭 conduitSubscription 功能。

配置陷阱

如果代理属性 advisorySupport 被禁用，网络将无法按预期工作（它们无法动态响应新的消费者）。如果 advisorySupport 被禁用，那么唯一的选择是完全静态配置的网络。在下一节中阅读更多相关内容。

代理网络和通知

代理网络严重依赖通知消息，因为它们在后台用于表示对远程端新的消费者的兴趣。默认情况下，当网络连接器启动时，它会在以下主题上定义一个消费者 ActiveMQ.Advisory.Consumer.>（暂时忽略临时目的地）。这样，当消费者连接（或断开连接）到远程代理时，本地代理将收到通知，并将把它视为它必须处理的另一个消费者。

在小型网络和目的地和消费者数量较少的环境中，这都很好。但是，随着规模的扩大，默认模型（监听所有内容，共享所有内容）将无法很好地扩展。这就是为什么您可以使用多种方法来过滤将在代理之间共享的目的地。

动态网络

让我们从动态配置的网络开始。这意味着我们只希望在远程代理上存在消费者时才将消息发送到该代理。如果我们想将此行为限制在某些目的地上，我们将使用 dynamicallyIncludedDestinations，例如

<networkConnector uri="static:(tcp://host)">
  <dynamicallyIncludedDestinations>
    <queue physicalName="include.test.foo"/>
    <topic physicalName="include.test.bar"/>
  </dynamicallyIncludedDestinations>
</networkConnector>

在 5.6 之前的 ActiveMQ Classic 版本中，代理仍然会使用相同的通知过滤器，并表示对远程代理上的所有消费者感兴趣。实际过滤将在消息分派期间完成。在大型网络中，这是一种次优的解决方案，因为它会在代理上产生大量的“通知”流量和负载。从 5.6 版本开始，代理将自动创建适当的通知过滤器，并只表示对动态包含的目的地感兴趣。对于我们的示例，它将是“ActiveMQ.Advisory.Consumer.Queue.include.test.foo,ActiveMQ.Advisory.Consumer.Topic.include.test.bar”。这可以极大地改善复杂和高负载环境中网络的行为。

在旧的代理版本中，我们可以通过稍微复杂的配置来实现相同的效果。控制我们感兴趣的消费者的实际通知过滤器是通过 destinationFilter 连接器属性定义的。它的默认值为“>”，它与 "ActiveMQ.Advisory.Consumer." 前缀连接在一起。因此，要实现相同的效果，我们需要执行以下操作

<networkConnector uri="static:(tcp://host)" destinationFilter="Queue.include.test.foo,ActiveMQ.Advisory.Consumer.Topic.include.test.bar">
  <dynamicallyIncludedDestinations>
    <queue physicalName="include.test.foo"/>
    <topic physicalName="include.test.bar"/>
  </dynamicallyIncludedDestinations>
</networkConnector>

注意，第一个目的地没有前缀，因为它已经隐含。设置和维护起来稍微复杂一些，但它会起作用。如果您使用的是 5.6 或更高版本的代理，只需使用 dynamicallyIncludedDestinations 包含所需的目的地即可。

这也解释了为什么如果您在代理上关闭通知支持，动态网络将无法工作。在这种情况下，代理无法动态响应新的消费者。

纯静态网络

如果您希望完全保护代理免受远程代理上消费者的任何影响，或者如果您希望使用代理作为简单的代理，并将所有消息转发到远程端，无论是否存在消费者，您都应该考虑静态网络。

<networkConnector uri="static:(tcp://host)" staticBridge="true">
        <staticallyIncludedDestinations>
      		<queue physicalName="always.include.queue"/>
        </staticallyIncludedDestinations>
</networkConnector>

从 5.6 版本开始，staticBridge 参数可用，这意味着本地代理不会订阅远程代理上的任何通知主题，这意味着它对是否存在任何消费者不感兴趣。此外，您需要在 staticallyIncludedDestinations 中添加一个目的地列表。这将与在目的地上的额外消费者具有相同的效果，因此消息也将被转发到远程代理。由于 ActiveMQ Classic 的早期版本中没有 staticBridge 参数，您可以通过将 destinationFilter 设置为监听未使用的通知主题来欺骗代理，例如

<networkConnector uri="static:(tcp://host)" destinationFilter="NO_DESTINATION">
        <staticallyIncludedDestinations>
      		<queue physicalName="always.include.queue"/>
        </staticallyIncludedDestinations>
</networkConnector>

如果按此配置，代理将尝试监听 ActiveMQ.Advisory.Consumer.NO_DESTINATION 上的新消费者，该主题永远不会有消息，因此它将受到保护，不会收到有关远程代理消费者的信息。

动态网络和虚拟目的地（5.13.0 新功能）

如上所述，可以将代理网络配置为仅在远程代理上存在消费者时才将消息发送到该代理。但是，让我们考虑一些在使用 虚拟目的地 时动态流如何发生的情况。

虚拟目的地消费者和复合目的地

以下是一个将两个代理联网在一起的示例。本地代理包含使用 dynamicallyIncludedDestination 配置的网络连接器，而远程代理使用复合主题进行配置

本地代理

<networkConnector uri="static:(tcp://host)">
	<dynamicallyIncludedDestinations>
    	<topic physicalName="include.bar"/>
	</dynamicallyIncludedDestinations>
</networkConnector>

远程代理

<compositeTopic name="include.bar" forwardOnly="false">
    <forwardTo>
        <queue physicalName="include.bar.forward" />
    </forwardTo>
</compositeTopic >

在这个示例中，让我们考虑远程代理上队列 include.bar.forward 上的单个消费者。如果直接将消息发送到远程代理上的主题 include.bar，它将被转发到队列 include.bar.forward，并且消费者将收到它。但是，如果将消息发布到本地代理上的相同主题，则此消息将不会被转发到远程代理。

消息没有被转发，因为在 queue include.bar.forward 上的消费者不会被检测为本地代理中的 dynamicallyIncludedDestinations 列表的一部分。除非在原始主题（在本例中为 include.bar）上存在消费者，否则消息将不会被转发到远程代理。可以通过配置本地代理以监听虚拟目的地订阅来解决此问题。

首先，我们需要配置远程代理以在消费者订阅与虚拟目的地匹配的目的地时发送通知消息。在本例中，匹配是通过使用目的地过滤器内部确定的，该过滤器确定一个目的地是否转发到另一个目的地。要启用此功能，请在远程代理上将属性 useVirtualDestSubs 设置为 true

远程代理

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://activemq.org/config/1.0">

  <broker name="remoteBroker" useVirtualDestSubs="true">  
     .....
  </broker>

</beans>

接下来，本地代理上的网络连接器需要配置为监听新的通知消息，方法是将属性 useVirtualDestSubs 设置为 true

本地代理

<networkConnector uri="static:(tcp://host)" useVirtualDestSubs="true">
  <dynamicallyIncludedDestinations>
    <topic physicalName="include.bar"/>
  </dynamicallyIncludedDestinations>
</networkConnector>

现在，如果消费者订阅了远程代理上的队列 include.bar.forward，本地代理将转发发送到主题 include.bar. 的消息。

在目的地创建时的虚拟目的地消费者

现在让我们考虑上面的用例，其中存在相同的复合主题，但队列上没有消费者。
  远程代理

<compositeTopic name="include.bar" forwardOnly="false">
    <forwardTo>
        <queue physicalName="include.bar.forward" />
    </forwardTo>
</compositeTopic >

远程代理上配置了复合主题，而本地代理与之联网。

即使我们启用了 useVirtualDestSubs，除非消费者订阅了转发的队列，否则消息将不会被转发到远程代理。如果没有消费者，消息将被丢弃，因为它们被发送到本地代理上的主题 (include.bar)。在这种情况下，希望基于存在虚拟目的地的消息被转发，该虚拟目的地转发到

• 一个或多个队列；或
• 具有持久订阅的主题。

这两个条件都被视为对本地代理发出消息需求，尽管这些目的地上没有活动消费者。

远程代理

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://activemq.org/config/1.0">

  <broker name="remoteBroker" useVirtualDestSubs="true" useVirtualDestSubsOnCreation="true">  
     .....
  </broker>

</beans>

使用此配置，当队列 include.bar.forward 被创建时，虚拟目的地消费者通知将被发送到本地代理，以便它知道根据存在转发到队列的复合主题来转发消息。

复合目的地消费者和虚拟主题

以上示例展示了如何配置复合目的地，但虚拟主题也可以工作。在下面的示例中，远程代理上虚拟主题的队列上的消费者现在将产生需求，并且消息将通过网络从本地代理发送。

本地代理

<networkConnector uri="static:(tcp://host)">
  <dynamicallyIncludedDestinations>
    <topic physicalName="VirtualTopic.>"/>
  </dynamicallyIncludedDestinations>
</networkConnector>

远程代理

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://activemq.org/config/1.0">

  <broker name="remoteBroker" useVirtualDestSubs="true" >  
     .....
  </broker>

</beans>

使用 NetworkConnector 属性的示例配置

这是 Broker 示例配置的一部分

<networkConnectors>
      <networkConnector uri="static:(tcp://127.0.0.1:61617)"
         name="bridge"
         conduitSubscriptions="true"
         decreaseNetworkConsumerPriority="false">
      	<dynamicallyIncludedDestinations>
      		<queue physicalName="include.test.foo"/>
      		<topic physicalName="include.test.bar"/>
      	</dynamicallyIncludedDestinations>
      	<excludedDestinations>
      		<queue physicalName="exclude.test.foo"/>
      		<topic physicalName="exclude.test.bar"/>
      	</excludedDestinations>
        <staticallyIncludedDestinations>
      		<queue physicalName="always.include.queue"/>
      		<topic physicalName="always.include.topic"/>
      	</staticallyIncludedDestinations>
      </networkConnector>
    </networkConnectors>

请注意，目前 excludedDestinations 属性不会影响 staticallyIncludedDestinations。

可以在两个 Broker 之间设置多个网络连接器。每个网络连接器使用一个底层传输连接，因此您可能希望这样做以提高吞吐量或获得更灵活的配置。

例如，如果您使用分布式队列，您可能希望在网络上对队列接收器进行等效加权，但前提是接收器处于活动状态 - 例如。

<networkConnectors>
      <networkConnector uri="static:(tcp://127.0.0.1:61617)"
         name="queues_only"
         conduitSubscriptions="false"
         decreaseNetworkConsumerPriority="false">
      	<excludedDestinations>
      		<topic physicalName=">"/>
      	</excludedDestinations>
      </networkConnector>
    </networkConnectors>

注意：您只能在 excludedDestinations 和 dynamicallyIncludedDestinations 属性中使用 通配符。
注意：如果您在整个网络中使用持久主题订阅者，请勿更改桥接器的名称或 Broker 的名称。在内部，ActiveMQ Classic 使用网络名称和 Broker 名称来构建一个唯一的且可重复的持久订阅者名称。

卡住的消息

卡住的消息可能有多种原因，以下部分旨在提供设置建议，以帮助解决此问题。

禁用 replayWhenNoConsumers（版本 5.6）

有一个目标策略允许队列通过配置 conditionalNetworkBridgeFilterFactory（其中 replayWhenNoConsumers=true）来实现此行为。 conditionalNetworkBridgeFilterFactory 提供了一个可选的 replayDelay，它基于进入 Broker 的时间。

    <destinationPolicy>
      <policyMap>
        <policyEntries>
          <policyEntry queue="TEST.>" enableAudit="false">
            <networkBridgeFilterFactory>
              <conditionalNetworkBridgeFilterFactory replayWhenNoConsumers="true"/>
            </networkBridgeFilterFactory>
          </policyEntry>
        </policyEntries>
      </policyMap>
    </destinationPolicy>

注意：当使用 replayWhenNoConsumers=true（版本 < 5.9）时，还需要禁用光标重复检测，方法是使用 enableAudit=false，因为光标可能会将重放的消息标记为重复消息（具体取决于播放和通过网络桥接重放这些消息的时间窗口）。这个问题在 这篇文章 中进行了详细解释。

激活 decreaseNetworkConsumerPriority

在 networkConnector 中设置 decreaseNetworkConsumerPriority=”true” 以限制咨询和优先本地消费者的数量。（有关 AMQ-7316 的更多信息）

<networkConnectors>
  <networkConnector uri="static:(tcp://mybroker:61616)" decreaseNetworkConsumerPriority="true" />
</networkConnectors>

如果您使用消息组和消费者池，请增加预取大小

如果启用了消息组，一个组可能会阻止队列的复制（如果该组中的消息大小大于配置的预取大小），提供足够的预取大小以确保所有消费者线程都可以消耗其分配的组。

增加 TTL

Broker 网络的 TTL 设置（networkTTL、messageTTL 和 consumerTTL）应该足够高，以允许消息在 Broker 之间交换多次，以防消费者反复失败。

限制网络消费者

conditionalNetworkBridgeFilterFactory 工厂允许为目标指定速率限制，以便可以限制网络消费者。网络消费者的预取在很大程度上被网络消费者通常非常快地确认消息的事实所抵消，因此即使预取量很低，优先级也很低，网络消费者也可以使一个速度适中的本地消费者陷入饥饿状态。限制提供了对此的补救措施。