activemq 双机——ActiveMQ集群Pure Master Slave

文章来源http://hi.baidu.com/minordragon/blog/item/6752c21fcad2b211314e15e7.html

如下：

解决单点故障：Pure Master Slave

1.         Master：给broker取个名字，修改其持久化KAHADB文件

2.         Slave：给broker取个名字，修改其持久化KAHADB文件，需要配置Master的地址和端口

3.         一个Master只能带一个Slave

4.         Master工作期间，会将消息状况自动同步到Slave

5.         Master一旦崩溃，Slave自动接替其工作，已发送并尚未消费的消息继续有效

6.         Slave接手后，必须停止Slave才能重启先前的Master

     MQ配置

1.         Master：首先复制 $activemq_dir$/conf/activemq.xml，并改名为：pure_master.xml，修改文件(192.168.0.20)

1)         <broker brokerName="pure_master" …

2)         <kahaDB directory="${activemq.base}/data/kahadb_pure_master "/>

3)         <transportConnector name="openwire" uri="tcp://0.0.0.0:61616"/>

2.         Slave：首先复制 $activemq_dir$/conf/activemq.xml，并改名为：pure_slave.xml，修改文件(192.168.0.86)

1)         <broker brokerName="pure_slave" masterConnectorURI="tcp://192.168.0.20:61616"  //master的ip

shutdownOnMasterFailure="false" …

2)         <kahaDB directory="${activemq.base}/data/kahadb_pure_slave "/>

3)         <transportConnector name="openwire" uri="tcp://0.0.0.0:61616"/>

3.         首先启动Master，启动完毕后在另一个Shell启动Slave，Slave启动后，可以看到Master那个Shell中显示已经Attach上了Slave

1)         启动Master：$activemq_dir$\bin>activemq xbean:file:../conf/pure_master.xml

2)         启动Slave：$activemq_dir$\bin>activemq xbean:file:../conf/pure_slave.xml

     JAVA测试：队列

1.         生产者

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

         ConnectionFactory cf = new ActiveMQConnectionFactory("failover:(tcp://0.0.0.0:61616,tcp://0.0.0.0:61617)");

         Connection conn = cf.createConnection();

         conn.start();

         Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

         Queue qq = new ActiveMQQueue("qq1");

         MessageProducer prod = sess.createProducer(qq);

         Message msg = null;

        

         Scanner scan = new Scanner(System.in);

         String str = scan.next();

         while(true) {

                   msg = sess.createTextMessage(str);

                   prod.send(msg);

                   if(str.equals("exit")) {

                            break;

                   }

                   str = scan.next();

         }

         conn.close();

}

2.         消费者

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

         ConnectionFactory cf = new ActiveMQConnectionFactory("failover:( tcp://0.0.0.0:61616,tcp://0.0.0.0:61617)");

         Connection conn = cf.createConnection();      

         conn.start();

         Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

         Queue qq = new ActiveMQQueue("qq1");

         MessageConsumer cs = sess.createConsumer(qq);

 

         TextMessage msg = (TextMessage)cs.receive();

         String str = msg.getText();

         while(true) {

                   System.out.println("receive msg:\t"+msg.getText());

                   if(str.equals("exit")) {

                            break;

                   }

                   msg = (TextMessage)cs.receive();

                   str = msg.getText();

         }

         conn.close();

}

     测试步骤

1.         启动生产者，发送几条消息

2.         启动消费者，可看到接收到的消息

3.         关闭消费者

4.         生产者继续发送几条消息—消息A

5.         停止Master（可看到生产者端显示连接到Slave（tcp://0.0.0.0:61617）了）

6.         生产者继续发送几条消息—消息B

7.         启动消费者

8.         消费者接收了消息A和消息B，可见Slave接替了Master的工作，而且储存了之前生产者经过Master发送的消息

     结论

Pure Master Slave模式实现方式简单，可以实现消息的双机热备功能；队列可以实现消息的异步和点对点发送