ActiveMQ部署方案分析对比.docx

ActiveMQ集群部署方式对比

ActiveMQ集群部署方式包括如下几种：

Pure Master Slave方式
Shared Filesystem Master-Slave方式
Shared Database Master-Slave方式
Replicated LevelDB Store方式
Broker-Cluster部署方式
模式介绍
传统的主从模式配置: 只能是Master提供服务，Slave是实时地备份Master的数据，以保证消息的可靠性。当Master失效时，Slave会自动升级为Master，客户端会自动连接到Slave上工作
利用共享文件系统做ActiveMQ集群，是基于ActiveMQ的默认数据库kahaDB完成的，kahaDB的底层是文件系统。这种方式的集群，Slave的个数没有限制，哪个ActiveMQ实例先获取共享文件的锁，那个实例就是Master，其它的ActiveMQ实例就是Slave，当当前的Master失效，其它的Slave就会去竞争共享文件锁，谁竞争到了谁就是Master。
和Shared File Sysytem Master Slave模式的原理是一样的，只是把共享文件系统换成了共享数据库
这种主备方式是ActiveMQ5.9以后才新增的特性，使用ZooKeeper协调选择一个node作为master。被选择的master broker node开启并接受客户端连接。
主要是通过network of Brokers在多个ActiveMQ实例之间进行消息的路由，可以解决服务负载问题。
关注度
低
高
低
高
低
利用率
低
5.9版本后取消
高
低
高
低
单点故障
解决，但是只支持一次故障。
解决
解决
解决
不解决
高可用
否
否
否
否
是
评价
老版本的部署方式，在5.9版本后已经取消了这么部署模式，可以支持一次故障。
是目前较流行的部署方式，使用active自带的db模式。可以使用nas文件系统方式实现多机部署。
只是将文件系统模式改成了数据库模式，这种模式比文件系统模式简单，但是一旦数据库失效，则所有的activemq实例都会失效。
这种是5.9版本后增加的新的特性，属于典型的互联网架构设计，引入zookeeper，如果有现成的zookeeper集群，则部署简单，否则，zookeeper出现故障后，会影响activemq集群。
实现了高可用性，但是没有解决单点问题。一旦实例出现故障，则会造成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>
            <!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
<!-- add &amp;wireFormat.maxInactivityDuration=0 -->
            <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&amp;wireformat.maxFrameSize=104857600&amp;wireFormat.maxInactivityDuration=0" discoveryUri="multicast://default"/>
            <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&amp;wireformat.maxFrameSize=104857600&amp;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>
                <!--
                    Enable this connector if you wish to use https with web console
                -->
                <!--
                <bean id="SecureConnector" class="org.eclipse.jetty.server.ssl.SslSelectChannelConnector">
                    <property name="port" value="8162" />
                    <property name="keystore" value="file:${activemq.conf}/broker.ks" />
                    <property name="password" value="password" />
                </bean>
                -->
            </list>
</property>

ActiveMQ B
1.activemq.xml修改监听端口:

<transportConnectors>
            <!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
<!-- add &amp;wireFormat.maxInactivityDuration=0 -->
            <transportConnector name="openwire" uri="tcp://0.0.0.0:61617?maximumConnections=1000&amp;wireformat.maxFrameSize=104857600&amp;wireFormat.maxInactivityDuration=0" discoveryUri="multicast://default"/>
            <transportConnector name="amqp" uri="amqp://0.0.0.0:5673?maximumConnections=1000&amp;wireformat.maxFrameSize=104857600&amp;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>
                <!--
                    Enable this connector if you wish to use https with web console
                -->
                <!--
                <bean id="SecureConnector" class="org.eclipse.jetty.server.ssl.SslSelectChannelConnector">
                    <property name="port" value="8162" />
                    <property name="keystore" value="file:${activemq.conf}/broker.ks" />
                    <property name="password" value="password" />
                </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();   
    }   

    // 关闭连接  
    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();   
        }   
    }   
} 

3.Main

import javax.jms.JMSException;

public class Test {   

    /**  
     * @param args  
     */  
    public static void main(String[] args) throws JMSException, Exception {

        ConsumerTool consumer = new ConsumerTool();   
        ProducerTool producer = new ProducerTool();   
        // 开始监听   
        consumer.consumeMessage();   

        // 延时500毫秒之后发送消息   
        Thread.sleep(500);   
        producer.produceMessage("Hello, world!");   
        producer.close();   

        // 延时500毫秒之后停止接受消息   
        Thread.sleep(500);   
        consumer.close();   

    }   
} 

ActiveMQ A 启动界面：

ActiveMQ B 启动界面：

AMQ A先启动，先锁文件，当AMQ B启动是，不能锁文件，但会不断的监听等待。

运行Java Test程序日志：

10:22:43.745 INFO  [] org.apache.activemq.transport.failover.FailoverTransport - Successfully connected to tcp://172.16.30.11:61616
Consumer:->Begin listening...
10:22:45.623 INFO  [] org.apache.activemq.transport.failover.FailoverTransport - Successfully connected to tcp://172.16.30.11:61616?wireFormat.maxInactivityDuration=0
Producer:->Sending message: Hello, world!
Producer:->Message sent complete!
Producer:->Closing connection
Consumer:->Received: Hello, world!
Consumer:->Closing connection

ActiveMQ A 管理界面：

异常处理：

配置好ActiveMQ后，前几次都启动成功。有一天启动时发现启动不成功，查看报错日志发现出现如下提示：
Failed to start Apache ActiveMQ (localhost, ID:*-PC-*-*-0:1). Reason: java.io.IOException: Transport Connector could not be registered in JMX: Failed to bind to server socket: tcp://0.0.0.0:61616?maximumConnections=1000&wireformat.maxFrameSize=104857600 due to: .BindException: Address already in use: JVM_Bind。

1.先去查看是不是端口被占用，用netstat -ano命令查看端口使用情况，发现没