消息队列：ActiveMQ：ActiveMQ的集群与负载均衡

消息队列：ActiveMQ：ActiveMQ的集群与负载均衡1消息队列基础1.1ActiveMQ简介ActiveMQ是一个开源的消息中间件，基于Java开发，支持多种消息协议，如AMQP、STOMP、MQTT等。它提供了高性能、高可用性以及易于使用的特性，使其成为企业级应用中消息传递的首选解决方案。ActiveMQ支持多种消息模式，包括点对点(Point-to-Point,P2P)和发布/订阅(Publish/Subscribe,Pub/Sub)，能够满足不同场景下的消息传递需求。1.2消息队列的工作原理消息队列是一种用于在分布式系统中进行消息传递的机制。它允许应用程序将消息发送到队列中，然后由其他应用程序从队列中读取消息。这种机制可以提高系统的解耦性、可靠性和可扩展性。1.2.1点对点(P2P)模式在P2P模式下，消息被发送到队列中，每个消息只能被一个消费者消费。一旦消息被消费，它就会从队列中移除。示例代码importorg.apache.activemq.ActiveMQConnectionFactory;

importjavax.jms.*;

publicclassP2PConsumer{

publicstaticvoidmain(String[]args)throwsJMSException{

//创建连接工厂

ConnectionFactoryconnectionFactory=newActiveMQConnectionFactory("tcp://localhost:61616");

//创建连接

Connectionconnection=connectionFactory.createConnection();

//启动连接

connection.start();

//创建会话

Sessionsession=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);

//创建队列

Queuequeue=session.createQueue("P2PQueue");

//创建消息消费者

MessageConsumerconsumer=session.createConsumer(queue);

//接收消息

TextMessagemessage=(TextMessage)consumer.receive();

//输出消息内容

System.out.println("Receivedmessage:"+message.getText());

//关闭资源

consumer.close();

session.close();

connection.close();

}

}1.2.2发布/订阅(Pub/Sub)模式在Pub/Sub模式下，消息被发送到主题中，所有订阅该主题的消费者都会接收到消息。这种模式适用于一对多的消息传递场景。示例代码importorg.apache.activemq.ActiveMQConnectionFactory;

importjavax.jms.*;

publicclassPubSubConsumer{

publicstaticvoidmain(String[]args)throwsJMSException{

//创建连接工厂

ConnectionFactoryconnectionFactory=newActiveMQConnectionFactory("tcp://localhost:61616");

//创建连接

Connectionconnection=connectionFactory.createConnection();

//启动连接

connection.start();

//创建会话

Sessionsession=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);

//创建主题

Topictopic=session.createTopic("PubSubTopic");

//创建消息消费者

MessageConsumerconsumer=session.createConsumer(topic);

//接收消息

TextMessagemessage=(TextMessage)consumer.receive();

//输出消息内容

System.out.println("Receivedmessage:"+message.getText());

//关闭资源

consumer.close();

session.close();

connection.close();

}

}1.3ActiveMQ在企业级应用中的角色ActiveMQ在企业级应用中扮演着消息中间件的角色，它能够处理大量的消息传递任务，提供消息的持久化、事务支持、消息优先级等功能，确保消息的可靠传递。此外，ActiveMQ还支持集群和负载均衡，能够提高系统的可用性和性能。1.3.1集群配置示例在ActiveMQ中，可以通过配置集群来提高系统的可用性和性能。以下是一个简单的集群配置示例：示例代码<beanid="broker"class="org.apache.activemq.broker.BrokerService">

<propertyname="brokerName"value="broker1"/>

<propertyname="dataDirectory"value="${activemq.data}/broker1"/>

<propertyname="transportConnectors">

<list>

<refbean="transportConnector"/>

</list>

</property>

<propertyname="clusterConnectors">

<list>

<refbean="clusterConnector"/>

</list>

</property>

</bean>

<beanid="transportConnector"class="org.apache.activemq.transport.nio.NioConnector">

<propertyname="uri"value="nio://:61616"/>

</bean>

<beanid="clusterConnector"class="org.apache.activemq.transport.nio.NioConnector">

<propertyname="uri"value="nio://broker2:61616"/>

</bean>1.3.2负载均衡示例ActiveMQ的负载均衡可以通过配置多个Broker实例，并使用负载均衡器（如NIO负载均衡器）来实现。以下是一个简单的负载均衡配置示例：示例代码<beanid="broker"class="org.apache.activemq.broker.BrokerService">

<propertyname="brokerName"value="broker1"/>

<propertyname="dataDirectory"value="${activemq.data}/broker1"/>

<propertyname="transportConnectors">

<list>

<refbean="transportConnector"/>

</list>

</property>

</bean>

<beanid="transportConnector"class="org.apache.activemq.transport.nio.NioConnector">

<propertyname="uri"value="nio://:61616"/>

<propertyname="loadBalancingPolicy">

<beanclass="org.apache.activemq.loadbalancer.NioLoadBalancingPolicy">

<propertyname="brokers">

<list>

<value>nio://broker1:61616</value>

<value>nio://broker2:61616</value>

</list>

</property>

</bean>

</property>

</bean>通过上述配置，ActiveMQ可以在多个Broker实例之间进行负载均衡，提高系统的处理能力和可用性。2消息队列：ActiveMQ：ActiveMQ的集群与负载均衡2.1ActiveMQ集群概念2.1.1集群的定义在计算机科学中，集群（Cluster）是指一组连接在一起的计算机，它们作为一个整体对外提供服务。集群技术主要用于提高系统的可用性、性能和可扩展性。在消息队列的场景下，集群可以确保即使单个节点失败，消息处理也不会中断，同时通过负载均衡，可以更高效地处理大量消息。2.1.2集群的优势高可用性：通过多个节点的冗余，确保即使部分节点故障，整个系统仍能继续运行。负载均衡：消息可以被分发到集群中的多个节点，避免单点过载，提高处理效率。可扩展性：随着需求的增长，可以通过增加集群中的节点来提升系统的处理能力。2.1.3ActiveMQ集群的实现方式ActiveMQ支持多种集群模式，包括：Master-Slave模式在Master-Slave模式下，一个节点作为Master，负责接收和处理消息，而其他节点作为Slave，通常处于待机状态。当Master节点发生故障时，其中一个Slave节点会接管Master的角色，确保服务的连续性。Broker-to-Broker模式Broker-to-Broker模式是通过多个Broker（消息代理）之间的直接通信来实现集群的。每个Broker都可以接收和处理消息，同时它们之间会同步消息状态，确保消息的一致性和高可用性。VirtualDestinations模式VirtualDestinations模式允许在集群中创建虚拟目的地，消息可以被发送到这些虚拟目的地，然后由集群中的Broker根据负载情况将消息分发到实际的物理目的地。代码示例：Broker-to-Broker模式//创建ActiveMQ连接工厂

ConnectionFactoryconnectionFactory=newActiveMQConnectionFactory("tcp://localhost:61616,tcp://localhost:61617");

//创建连接

Connectionconnection=connectionFactory.createConnection();

//启动连接

connection.start();

//创建会话

Sessionsession=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);

//创建目的地

Destinationdestination=session.createQueue("exampleQueue");

//创建消息生产者

MessageProducerproducer=session.createProducer(destination);

//创建消息

TextMessagemessage=session.createTextMessage("Hello,ActiveMQCluster!");

//发送消息

producer.send(message);

//关闭资源

session.close();

connection.close();在上述示例中，我们创建了一个连接到两个Broker的ConnectionFactory，这模拟了Broker-to-Broker集群的环境。通过这种方式，消息可以被发送到集群中的任何一个Broker，然后由集群内部机制确保消息的正确处理和分发。NetworkConnector模式NetworkConnector模式允许ActiveMQBroker通过网络连接器与其他Broker通信，实现消息的跨Broker传输。这种模式下，消息可以从一个Broker发送到另一个Broker，从而实现负载均衡和高可用性。代码示例：NetworkConnector模式<transportConnectors>

<transportConnectorname="openwire"uri="tcp://localhost:61616"/>

</transportConnectors>

<destinationInterceptors>

<destinationInterceptor>

<virtualDestinationInterceptor>

<virtualDestinationPrefix>virtual.</virtualDestinationPrefix>

<networkConnectors>

<networkConnectoruri="static:(tcp://localhost:61617)"/>

</networkConnectors>

</virtualDestinationInterceptor>

</destinationInterceptor>

</destinationInterceptors>在配置文件中，我们定义了一个networkConnector，它将消息从一个Broker（tcp://localhost:61616）转发到另一个Broker（tcp://localhost:61617）。通过这种方式，消息可以在集群中不同Broker之间流动，实现更灵活的负载均衡和高可用性。通过以上介绍，我们可以看到ActiveMQ提供了多种集群实现方式，每种方式都有其特定的使用场景和优势。在实际应用中，根据系统的具体需求选择合适的集群模式，可以显著提升消息处理的效率和系统的稳定性。3配置ActiveMQ集群3.1设置集群环境在设置ActiveMQ集群环境之前，我们首先需要理解ActiveMQ集群的基本概念。ActiveMQ集群允许多个Broker实例协同工作，以提供高可用性和负载均衡。集群中的每个Broker都可以接收和发送消息，而消息的持久化和复制则确保了即使某个Broker失败，消息也不会丢失。3.1.1环境准备安装ActiveMQ：确保在每台服务器上都安装了ActiveMQ。可以下载最新版本的ActiveMQ并解压到指定目录。配置JVM参数：为了确保集群的稳定性，需要在每台服务器上配置JVM参数，例如增加堆内存大小。网络配置：确保所有Broker之间网络畅通，可以互相通信。3.1.2配置Broker实例在每个Broker实例的conf/activemq.xml文件中，我们需要进行以下配置：<!--activemq.xml-->

<transportConnectors>

<transportConnectorname="openwire"uri="tcp://:61616"/>

<transportConnectorname="cluster"uri="static:(tcp://00:61617,tcp://01:61617)"/>

</transportConnectors>

<destinationPolicy>

<destinationPolicy>

<queuephysicalName="QueueName"policyEntries>

<policyEntryqueue="QueueName"producerFlowControl="false"/>

</queue>

</destinationPolicy>

</destinationPolicy>

<systemUsage>

<memoryUsage>

<percentageOfJvmHeappercentage="60"/>

</memoryUsage>

</systemUsage>

<brokerxmlns="/schema/core"brokerName="BrokerName"dataDirectory="data">

<persistenceAdapter>

<kahaDBdirectory="kahadb"/>

</persistenceAdapter>

</broker>在上述配置中，transportConnectors定义了Broker的网络连接，cluster是集群通信的连接器。destinationPolicy用于控制消息的流向，systemUsage定义了系统资源的使用策略，而persistenceAdapter则用于配置消息的持久化。3.2配置Broker间通信为了实现Broker之间的通信，我们需要在conf/activemq.xml中配置集群通信的细节。这包括定义集群的连接器、消息的复制策略等。3.2.1配置集群连接器集群连接器cluster用于定义Broker之间如何通信。在配置文件中，我们需要指定集群连接器的名称和URI，例如：<transportConnectors>

<transportConnectorname="cluster"uri="static:(tcp://00:61617,tcp://01:61617)"/>

</transportConnectors>这里的uri指定了集群中其他Broker的地址和端口。3.2.2配置消息复制策略ActiveMQ提供了多种消息复制策略，包括同步和异步复制。在conf/activemq.xml中，我们可以通过以下方式配置：<brokerxmlns="/schema/core"brokerName="BrokerName"dataDirectory="data">

<persistenceAdapter>

<kahaDBdirectory="kahadb"journalMaxFileLength="10240000"useJournalReplayForSlowConsumers="true"/>

</persistenceAdapter>

<networkConnectors>

<networkConnectorname="cluster"uri="static:(tcp://00:61617,tcp://01:61617)"networkTieBreaker="true"/>

</networkConnectors>

</broker>在上述配置中，networkConnectors定义了集群中Broker之间的网络连接，networkTieBreaker用于在多个Broker同时可用时选择一个Broker。3.3实现消息持久化消息持久化是消息队列系统中的一个重要特性，它确保了即使Broker重启或失败，消息也不会丢失。在ActiveMQ中，我们可以通过配置persistenceAdapter来实现消息的持久化。3.3.1使用KahaDBKahaDB是ActiveMQ默认的消息持久化存储。在conf/activemq.xml中，我们可以通过以下方式配置KahaDB：<persistenceAdapter>

<kahaDBdirectory="kahadb"journalMaxFileLength="10240000"useJournalReplayForSlowConsumers="true"/>

</persistenceAdapter>这里的directory指定了KahaDB数据存储的目录，journalMaxFileLength定义了日志文件的最大长度，而useJournalReplayForSlowConsumers则确保了慢消费者不会影响消息的持久化。3.3.2使用LevelDB除了KahaDB，ActiveMQ还支持LevelDB作为持久化存储。配置LevelDB的方式如下：<persistenceAdapter>

<levelDBdirectory="leveldb"/>

</persistenceAdapter>这里的directory指定了LevelDB数据存储的目录。3.3.3实例代码下面是一个使用Java代码发送和接收持久化消息的例子：importorg.apache.activemq.ActiveMQConnectionFactory;

publicclassPersistentMessageExample{

publicstaticvoidmain(String[]args)throwsException{

//创建连接工厂

ActiveMQConnectionFactoryconnectionFactory=newActiveMQConnectionFactory("tcp://localhost:61616");

//创建连接

Connectionconnection=connectionFactory.createConnection();

//启动连接

connection.start();

//创建会话

Sessionsession=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);

//创建队列

Destinationdestination=session.createQueue("PersistentQueue");

//创建消息生产者

MessageProducerproducer=session.createProducer(destination);

//设置消息持久化

producer.setDeliveryMode(DeliveryMode.PERSISTENT);

//创建消息

TextMessagemessage=session.createTextMessage("Hello,PersistentMessage!");

//发送消息

producer.send(message);

//创建消息消费者

MessageConsumerconsumer=session.createConsumer(destination);

//接收消息

TextMessagereceivedMessage=(TextMessage)consumer.receive();

//输出接收到的消息

System.out.println("Receivedmessage:"+receivedMessage.getText());

//关闭资源

consumer.close();

session.close();

connection.close();

}

}在这个例子中，我们创建了一个连接到本地ActiveMQBroker的连接，然后创建了一个会话和队列。通过设置producer.setDeliveryMode(DeliveryMode.PERSISTENT)，我们确保了发送的消息会被持久化。最后，我们创建了一个消费者来接收并输出持久化消息。通过上述步骤，我们可以成功地配置ActiveMQ集群，实现Broker之间的通信，并确保消息的持久化。这为构建高可用和高性能的消息队列系统提供了基础。4负载均衡原理4.1负载均衡的定义负载均衡（LoadBalancing）是一种将工作负载（如网络流量、计算任务、数据处理等）分散到多个计算资源（如服务器、网络设备、数据库等）上的技术。其主要目的是优化资源使用、提高系统响应速度、增强系统的可靠性和可扩展性。在消息队列系统中，负载均衡尤为重要，因为它可以确保消息的高效处理和系统的稳定运行。4.2负载均衡在ActiveMQ中的应用4.2.1ActiveMQ集群ActiveMQ支持多种集群模式，其中一种是基于负载均衡的集群。在ActiveMQ集群中，多个ActiveMQ实例通过网络连接，形成一个逻辑上的消息队列系统。当客户端发送消息到集群时，消息会被分发到集群中的不同节点上，这样可以提高消息处理的效率和系统的可用性。4.2.2负载均衡策略ActiveMQ的负载均衡策略通常包括以下几种：轮询（RoundRobin）：这是最简单的负载均衡策略，按照顺序将消息发送到集群中的每个节点。例如，如果有三个节点，那么第一个消息会被发送到第一个节点，第二个消息会被发送到第二个节点，以此类推，直到所有节点都被轮询过，然后再次从第一个节点开始。最少消息（LeastMessages）：这种策略会将消息发送到当前消息最少的节点，以确保所有节点的负载均衡。最少任务（LeastTasks）：与最少消息策略类似，但考虑的是节点上正在处理的任务数量，而不是消息数量。随机（Random）：将消息随机发送到集群中的任意一个节点，这种策略简单但可能不会达到最佳的负载均衡效果。基于权重（Weighted）：每个节点可以被赋予一个权重，消息将根据节点的权重进行分发，权重高的节点将接收更多的消息。4.2.3示例：配置ActiveMQ集群的负载均衡策略以下是一个配置ActiveMQ集群使用最少消息负载均衡策略的示例。假设我们有两个ActiveMQ节点，分别配置为node1和node2。<!--在ActiveMQ的配置文件中，定义集群策略-->

<transportConnectors>

<transportConnectorname="openwire"uri="tcp://localhost:61616?jmxPort=1099&wireFormat.maxFrameSize=10000000&wireFormat.cacheEnabled=false&wireFormat.cacheSize=10000&wireFormat.maxInactivityDurationBeforeConnectionLoss=30000&wireFormat.maxInactivityDurationInitalValue=30000&wireFormat.optimizeFrameProcessing=true&wireFormat.sendFrameSizeHeader=true&wireFormat.sendInactivityMonitorInfoHeader=true&wireFormat.sendTimeStamps=true&wireFormat.sendStackTraceOnException=true&wireFormat.sendConnectionLocators=true&wireFormat.sendConnectionId=true&wireFormat.sendBrokerInfo=true&wireFormat.sendNetworkConnectors=true&wireFormat.sendNetworkLocator=true&wireFormat.sendNetworkSubscription=true&wireFormat.sendNetworkSubscriptionDurable=true&wireFormat.sendNetworkSubscriptionNoLocal=true&wireFormat.sendNetworkSubscriptionAck=true&wireFormat.sendNetworkSubscriptionAckInterval=1000&wireFormat.sendNetworkSubscriptionAckSize=100&wireFormat.sendNetworkSubscriptionAckBatchSize=100&wireFormat.sendNetworkSubscriptionAckBatchInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThreshold=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSize=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThreshold=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSize=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThreshold=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSize=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThreshold=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSize=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThreshold=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSize=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThreshold=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSize=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThreshold=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSize=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThreshold=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSize=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThreshold=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSize=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThreshold=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSize=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThreshold=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSize=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThreshold=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSize=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThreshold=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSize=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThreshold=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSize=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThreshold=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSize=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThreshold=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSize=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThreshold=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSize=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThreshold=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSize=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThreshold=100&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdInterval=1000&wireFormat.sendNetworkSubscriptionAckBatchSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeThresholdSizeT