ActiveMQ 叢集配置

構建高可用的ActiveMQ系統在生產環境中是非常重要的，單點的ActiveMQ作為企業應用無法滿足高可用和叢集的需求，所以ActiveMQ提供了master-slave、broker cluster等多種部署方式，但通過分析多種部署方式之後我認為需要將兩種部署方式相結合才能滿足我們公司分散式和高可用的需求，所以後面就重點將解如何將兩種部署方式相結合。

自從activemq5.9.0開始，activemq的叢集實現方式取消了傳統的Pure Master Slave方式，增加了基於zookeeper+leveldb的實現方式，其他兩種方式：目錄共享和資料庫共享依然存在。

1、Master-Slave部署方式

1)、Shared Filesystem Master-Slave方式
2)、Shared Database Master-Slave方式
3)、Replicated LevelDB Store方式

第一種方案同樣支援N個AMQ例項組網，但由於他是基於kahadb儲存策略，亦可以部署在分散式檔案系統上，應用靈活、高效且安全。

第二種方案與shared filesystem方式類似，只是共享的儲存介質由檔案系統改成了資料庫而已，支援N個AMQ例項組網，但他的效能會受限於資料庫；

第三種方案是ActiveMQ5.9以後才新增的特性，使用ZooKeeper協調選擇一個node作為master。被選擇的master broker node開啟並接受客戶端連線。

其他node轉入slave模式，連線master並同步他們的儲存狀態。slave不接受客戶端連線。所有的儲存操作都將被複制到連線至Master的slaves。
如果master死了，得到了最新更新的slave被允許成為master。fialed node能夠重新加入到網路中並連線master進入slave mode。所有需要同步的disk的訊息操作都將等待儲存狀態被複制到其他法定節點的操作完成才能完成。所以，如果你配置了replicas=3，那麼法定大小是(3/2)+1=2. Master將會儲存並更新然後等待 (2-1)=1個slave儲存和更新完成，才彙報success。至於為什麼是2-1，熟悉Zookeeper的應該知道，有一個node要作為觀擦者存在。

單一個新的master被選中，你需要至少保障一個法定node線上以能夠找到擁有最新狀態的node。這個node將會成為新的master。因此，推薦執行至少3個replica nodes，以防止一個node失敗了，服務中斷。

Shared Filesystem Master-Slave方式

shared filesystem Master-Slave部署方式主要是通過共享儲存目錄來實現master和slave的熱備，所有的ActiveMQ應用都在不斷地獲取共享目錄的控制權，哪個應用搶到了控制權，它就成為master。

多個共享儲存目錄的應用，誰先啟動，誰就可以最早取得共享目錄的控制權成為master，其他的應用就只能作為slave。

Apache ActiveMQ單點基本配置的原配置內容：

<persistenceAdapter>
            <kahaDB directory="${activemq.data}/kahadb"/>
</persistenceAdapter>

SharedFile System Master Slave 修改為：

<persistenceAdapter>
             <kahaDB directory="D:\\ActiveMQ Cluster\\shareBrokerData" enableIndexWriteAsync="true" enableJournalDiskSyncs="false"/>
</persistenceAdapter>

在D:\\ActiveMQ Cluster目錄先建立shareBrokerData資料夾。

注意：

1.前面提到如果在一臺裝置上部署多個AMQ，需要修改對應埠號，如AMQ對外的監聽埠61616和jetty的監聽埠8161等。
2.如果多套AMQ部署在不同的裝置上，這裡的directory應該指向一個遠端的系統目錄（分散式檔案系統）
3.客戶端通過failover方式進行連線，多個AMQ例項地址使用英文逗號隔開，當某個例項斷開時會自動重連，但如果所有例項都失效，failover預設情況下會無限期的等待下去，不會有任何提示。

下面為在一臺裝置上部署兩個AMQ示例：
ActiveMQ A
1.activemq.xml修改監聽埠:

<transportConnectors>
            

            <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireformat.maxFrameSize=104857600&wireFormat.maxInactivityDuration=0" discoveryUri="multicast://default"/>
            <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&wireformat.maxFrameSize=104857600&wireFormat.maxInactivityDuration=0"/>

</transportConnectors>

2.jetty.xml修改監聽埠:

<property name="connectors">
            <list>
                <bean id="Connector" class="org.eclipse.jetty.server.nio.SelectChannelConnector">
                    <property name="port" value="8166" />
                </bean>
                
                
            </list>
</property>

ActiveMQ B
1.activemq.xml修改監聽埠:

<transportConnectors>
            

            <transportConnector name="openwire" uri="tcp://0.0.0.0:61617?maximumConnections=1000&wireformat.maxFrameSize=104857600&wireFormat.maxInactivityDuration=0" discoveryUri="multicast://default"/>
            <transportConnector name="amqp" uri="amqp://0.0.0.0:5673?maximumConnections=1000&wireformat.maxFrameSize=104857600&wireFormat.maxInactivityDuration=0"/>

</transportConnectors>

2.jetty.xml修改監聽埠:

<property name="connectors">
            <list>
                <bean id="Connector" class="org.eclipse.jetty.server.nio.SelectChannelConnector">
                    <property name="port" value="8167" />
                </bean>
                
                
            </list>
</property>

Java測試程式程式碼：
1.Producer:

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class ProducerTool {

    private String subject = "TOOL.DEFAULT";

    private Destination destination = null;

    private Connection connection = null;

    private Session session = null;

    private MessageProducer producer = null;

    // 初始化
    private void initialize() throws JMSException, Exception {
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("failover:(tcp://172.16.30.11:61616?wireFormat.maxInactivityDuration=0,tcp://172.16.30.11:61617?wireFormat.maxInactivityDuration=0)");
        connection = connectionFactory.createConnection();
        session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
        destination = session.createQueue(subject);
        producer = session.createProducer(destination);
        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

    }

    // 傳送訊息
    public void produceMessage(String message) throws JMSException, Exception {
        initialize();
        TextMessage msg = session.createTextMessage(message);
        connection.start();
        System.out.println("Producer:->Sending message: " + message);
        producer.send(msg);
        System.out.println("Producer:->Message sent complete!");
    }

    // 關閉連線
    public void close() throws JMSException {
        System.out.println("Producer:->Closing connection");
        if (producer != null)
            producer.close();
        if (session != null)
            session.close();
        if (connection != null)
            connection.close();
   }
}

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class ConsumerTool implements MessageListener {

    private String subject = "TOOL.DEFAULT";

    private Destination destination = null;

    private Connection connection = null;

    private Session session = null;

    private MessageConsumer consumer = null;

    // 初始化
    private void initialize() throws JMSException, Exception {
       ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("failover:(tcp://172.16.30.11:61616,tcp://172.16.30.11:61617)");
        connection = connectionFactory.createConnection();
        session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
        destination = session.createQueue(subject);
        consumer = session.createConsumer(destination);

    }

    // 消費訊息
    public void consumeMessage() throws JMSException, Exception {
        initialize();
        connection.start();

        System.out.println("Consumer:->Begin listening...");
        //
        consumer.setMessageListener(this);
        // Message message = consumer.receive();
    }

    // 關閉連線
    public void close() throws JMSException {
        System.out.println("Consumer:->Closing connection");
        if (consumer != null)
            consumer.close();
        if (session != null)
            session.close();
        if (connection != null)
            connection.close();
    }

    // 訊息處理函式
    public void onMessage(Message message) {
        try {
            if (message instanceof TextMessage) {
                TextMessage txtMsg = (TextMessage) message;
                String msg = txtMsg.getText();
                System.out.println("Consumer:->Received: " + msg);
            } else {
                System.out.println("Consumer:->Received: " + message);
            }
        } catch (JMSException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}

2.Consumer:

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class ConsumerTool implements MessageListener {

    private String subject = "TOOL.DEFAULT";

    private Destination destination = null;

    private Connection connection = null;

    private Session session = null;

    private MessageConsumer consumer = null;

    // 初始化
    private void initialize() throws JMSException, Exception {
       ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("failover:(tcp://172.16.30.11:61616,tcp://172.16.30.11:61617)");
        connection = connectionFactory.createConnection();
        session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
        destination = session.createQueue(subject);
        consumer = session.createConsumer(destination);

    }

    // 消費訊息
    public void consumeMessage() throws JMSException, Exception {
        initialize();
        connection.start();

        System.out.println("Consumer:->Begin listening...");
        //
        consumer.setMessageListener(this);
        // Message message = consumer.receive();
    }

    // 關閉連線

JMS學習十二（Spring+ActiveMQ叢集配置）

ActiveMQ 叢集配置

ActiveMQ叢集配置及使用

基於zookeeper的activemq的主從叢集配置

ActiveMQ的叢集配置

訊息佇列系列之ActiveMQ（JMS、叢集配置）

ActiveMQ基本配置

activeMq密碼配置 - 6

centos7下redis哨兵叢集配置

SSO CAS叢集配置

Apache +Jetty的負載均衡與叢集配置（下）

Apache +Jetty的負載均衡與叢集配置（上）

多節點高可用Eureka叢集配置與部署

Mariadb10 galera叢集配置

zookeeper叢集配置詳細教程

springboot上傳下載檔案（3）--java api 操作HDFS叢集+叢集配置

解決Centos6.4叢集配置ssh免密登陸仍然需要輸入密碼問題

redis cluster叢集配置步驟

Windows平臺下的Redis-Sentinel叢集配置

Kafka叢集配置---Windows版

ActiveMQ 叢集配置

相關推薦