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可以被處理