ActiveMQ的静态网络链接

--------------------------------------------------------------------

(1)ActiveMQ的networkConnector是什么

　　在某些情况下，需要多个ActiveMQ的Broker做集群，那么就涉及到Broker到Broker的通信，这个就称为ActiveMQ的networkConnector.

　　ActiveMQ的networkConnector默认是单向的，一个Broker在一端发送消息，另一个Broker在另一端接收消息，这就是所谓的"桥接"。ActiveMQ也支持双向链接，创建一个双向的通道对于两个Broker不仅发送消息而且也能从相同的通道接收消息，通常作为duplex connector来映射，如下：

(2)有两种配置Client到Broker的链接方式

　　第一种： Client通过Staticlly配置的方式去连接Broker(静态链接)

　　第二种: Client通过discover agent来dynamically的发现Brokers(动态链接)

Static networks:

　　Static networkConnector是用于创建一个静态的配置对于网络中的多个Broker,这种协议用于复合url,一个复合url包括多个url地址，格式如下:

　　static:(uri1,uri2,uri3, ...)?key=value

 <networkConnectors>

     <networkConnector name="local network" uri="static://(tcp://ip:prot,tcp://ip:port)"/>

 </networkConnectors>

　　在activemq.xml配置如下：

　下面启动两个Broker实例，一个端口:61616，另一个端口:61716，然后通过程序往61616端口的Broker发送数据，再从61716端口的Broker接收数据

JmsSend程序如下：

 import javax.jms.Connection;

 import javax.jms.ConnectionFactory;

 import javax.jms.Destination;

 import javax.jms.MessageProducer;

 import javax.jms.Session;

 import javax.jms.TextMessage;

 import org.apache.activemq.ActiveMQConnectionFactory;

 public class JmsSend {

     public static void main(String[] args) throws Exception {

         ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.81:61616");

         Connection connection = connectionFactory.createConnection();

         connection.start();

         Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);

         Destination destination = session.createQueue("my-queue");

         MessageProducer producer = session.createProducer(destination);

         for(int i = 0;i < 10;i++){

             TextMessage message = session.createTextMessage("message,1212 --->" + i);

             Thread.sleep(1000);

             //通过生产者发出消息

             producer.send(message);

         }

         session.commit();

         session.close();

         connection.close();

     }

 }

JmsReceiver程序如下:

 import javax.jms.Connection;

 import javax.jms.ConnectionFactory;

 import javax.jms.Destination;

 import javax.jms.MessageConsumer;

 import javax.jms.Session;

 import javax.jms.TextMessage;

 import org.apache.activemq.ActiveMQConnectionFactory;

 public class JmsReceiver {

     public static void main(String[] args) throws Exception {

         ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://192.168.1.81:61716");

         Connection connection =  cf.createConnection();

         connection.start();

         Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);

         Destination destination = session.createQueue("my-queue");

         MessageConsumer consumer = session.createConsumer(destination);

         int i = 0;

         while(i < 10){

             i++;

             TextMessage message = (TextMessage)consumer.receive();

             session.commit();

             System.out.println("接收到的消息是:"+message.getText());

         }

         session.close();

         connection.close();

     }

 }

运行结果:

(3)networkConnector配置的可用属性

　　1.name: 默认的bridge
　　2.dynamicOnly: 默认是false,如果为true,持久订阅被激活时才创建对应的网络持久订阅。默认是启动时激活
　　3.decreaseNetworkConsumerPriority: 默认是false。设定消费者优先权，如果为true,网络的消费者优先级降低为-5。如果为false,则默认跟本地消费者一样为0
　　4.networkTTL: 默认是1,网络中用于消息和订阅消费的broker数量
　　5.messageTTL: 默认是1，网络中用于消息的broker数量
　　6.consumerTTL: 默认是1，网络中用于消费的broker数量
　　7.conduitSubscriptions: 默认true,是否把同一个broker的多个consumer当做一个来处理（在做集群的时候如果有多个consumer，需要设置为false）
　　8.dynamicallyIncludedDestinations:默认为空，要包括的动态消息地址，类适于excludedDestinations,如：

 <dynamicallyIncludedDestinations>

     <queue physicalName="include.test.foo"/>

     <topic physicalName="include.test.bar"/>

 </dynamicallyIncludedDestinations>

　　9.staticallyIncludedDestinations:默认为空，要包括的静态消息地址。类似于excludedDestinations,如:

 <staticallyIncludedDestinations>

     <queue physicalName="always.include.queue"/>

 </staticallyIncludedDestinations>

　　10.excludedDestinations: 默认为空，指定排除的地址，示例如下:

 <networkConnectors>

 　　<networkConnector uri="static://(tcp://localhost:61617)" name="bridge" dynamicOnly="false" conduitSubscriptions="true"                                             decreaseNetworkConsumerPriority="false">

 　　<excludedDestinations>

 　　　　<queue physicalName="exclude.test.foo">

 　　　　<topic physicalName="exclude.test.bar">

 　　</excludedDestinations>

 　　<dynamicallyIncludedDestinations>

 　　　　<queue physicalName="include.test.foo"/>

 　　　　<topic physicalName="include.test.bar"/>

 　　</dynamicallyIncludedDestinations>

 　　<staticallyIncludedDestinations>

 　　　　<queue physicalName="always.include.queue"/>

 　　</staticallyIncludedDestinations>

 　　</networkConnector>

 </networkConnectors>

　　12. prefetchSize: 默认是1000，持有的未确认的最大消息数量，必须大于0，因为网络消费者不能自己轮询消息
　　13. suppressDuplicateQueueSubscriptions: 默认false,如果为true,重复的订阅关系一产生即被阻止
　　14. bridgeTempDestinations: 默认true,是否广播advisory messages来创建临时的destination
　　15. alwaysSyncSend: 默认false,如果为true,非持久化消息也将使用request/reply方式代替oneway方式发送到远程broker
　　16. staticBridge: 默认false,如果为true,只有staticallyIncludedDestinations中配置的destination可以被处理

巴特西

ActiveMQ的静态网络链接

最新文章

热门文章