ActiveMQ 集群配置 高可用

自從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。 

一、下載activeMQ解壓並復制3份

二、修改配置文件conf下面 activemq.xml

如下 ，3份文件broker都修改為一致  其中dataDirectory 是 數據存儲共享目錄地址

修改 kahaDB目錄地址為數據存儲共享目錄地址

 修改openwire  的tcp 連接地址端口分別為 61616 61626 61636 

二 、修改conf 下面的 jetty.xml文件，修改管理界面端口 8161 8162 8163 方便測試查看啟動的那一個activemq服務

 

 

 三、啟動  成功了一個后面的將阻塞等待獲取鎖 

四、測試

Receiver.java

package com.activemq.demo;

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.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class Receiver {
    public static void main(String[] args) {
        // ConnectionFactory ：連接工廠，JMS 用它創建連接
        ConnectionFactory connectionFactory;
        // Connection ：JMS 客戶端到JMS Provider 的連接
        Connection connection = null;
        // Session： 一個發送或接收消息的線程
        Session session;
        // Destination ：消息的目的地;消息發送給誰.
        Destination destination;
        // 消費者，消息接收者
        MessageConsumer consumer;
        connectionFactory = new ActiveMQConnectionFactory(
                "failover:(tcp://localhost:61616,tcp://localhost:61626,tcp://localhost:61636)");
        try {
            // 構造從工廠得到連接對象
            connection = connectionFactory.createConnection();
            // 啟動
            connection.start();
            // 獲取操作連接
            session = connection.createSession(Boolean.FALSE,
                    Session.AUTO_ACKNOWLEDGE);
            // 獲取session注意參數值xingbo.xu-queue是一個服務器的queue，須在在ActiveMq的console配置
            destination = session.createQueue("FirstQueue");
            consumer = session.createConsumer(destination);
            while (true) {
                //設置接收者接收消息的時間，為了便於測試，這里誰定為100s
                TextMessage message = (TextMessage) consumer.receive(100000);
                if (null != message) {
                    System.out.println("收到消息" + message.getText());
                } else {
                    break;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != connection)
                    connection.close();
            } catch (Throwable ignore) {
            }
        }
    }
}

　Sender.java

package com.activemq.demo;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class Sender {
    private static final int SEND_NUMBER = 5;

    public static void main(String[] args) {
        // ConnectionFactory ：連接工廠，JMS 用它創建連接
        ConnectionFactory connectionFactory;
        // Connection ：JMS 客戶端到JMS Provider 的連接
        Connection connection = null;
        // Session： 一個發送或接收消息的線程
        Session session;
        // Destination ：消息的目的地;消息發送給誰.
        Destination destination;
        // MessageProducer：消息發送者
        MessageProducer producer;
        // TextMessage message;
        // 構造ConnectionFactory實例對象，此處采用ActiveMq的實現jar
        connectionFactory = new ActiveMQConnectionFactory(
                "failover:(tcp://localhost:61616,tcp://localhost:61626,tcp://localhost:61636)");
        try {
            // 構造從工廠得到連接對象
            connection = connectionFactory.createConnection();
            // 啟動
            connection.start();
            // 獲取操作連接
            session = connection.createSession(Boolean.TRUE,
                    Session.AUTO_ACKNOWLEDGE);
            // 獲取session注意參數值xingbo.xu-queue是一個服務器的queue，須在在ActiveMq的console配置
            destination = session.createQueue("FirstQueue");
            // 得到消息生成者【發送者】
            producer = session.createProducer(destination);
            // 設置不持久化，此處學習，實際根據項目決定 ---- 集群此處必須持久化
         //   producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            // 構造消息，此處寫死，項目就是參數，或者方法獲取
            sendMessage(session, producer);
            session.commit();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != connection)
                    connection.close();
            } catch (Throwable ignore) {
            }
        }
    }

    public static void sendMessage(Session session, MessageProducer producer)
            throws Exception {
        for (int i = 1; i <= SEND_NUMBER; i++) {
            TextMessage message = session
                    .createTextMessage("ActiveMq 發送的消息" + i);
            // 發送消息到目的地方
            System.out.println("發送消息：" + "ActiveMq 發送的消息" + i);
            producer.send(message);
        }
    }
}