ActiveMQ主從配置

這種方式有個問題，activemq1有消息沒消費完但是突然宕機，雖然程序會自動連到activemq2。但是activemq1的消息只有等機器恢復后才會被消費。

1.啟動：我這里使用的是apache-activemq-5.13.3，是在windows下使用的，發現根據文檔說的雙擊activemq.bat啟動不了，那就只好使用命令啟動，CMD進入到apache-activemq-5.13.3\bin下，輸入activemqbat start。這樣就可以啟動了。

2.主從配置：第一個activemq解壓到apache-activemq-5.13.3，第二個解壓到apache-activemq-5.13.3-2

　　第一個activemq直接輸入命令啟動

　　第二個需要修改參數：a.打開apache-activemq-5.13.3-2\conf\activemq.xml，修改broker標簽里面的brokerName，不要和第一個相同就行

　　　　　　　　　　　　b.修改activemq.xml中的transportConnectors，刪除其他，只留一個openwire就行，修改uri里面的端口號

　　　　　　　　　　　　c.在transportConnectors上面添加（如果一會兒啟動的時候這里報錯，請手動敲打下面三行，不要復制）

　　　　　　　　　　　　　　<networkConnectors>

　　　　　　　　　　　　　　　　<networkConnector uri="static:(tcp://localhost:61616)" duplex="true"/>
　　　　　　　　　　　　　　</networkConnectors>

　　　　　　　　　　　　d.修改\conf\jetty.xml文件的115行，端口號隨便寫一個。（這里是jetty的訪問端口）

配置文件修改完成，啟動第一個activemq，啟動第二個activemq。

接下來是代碼中brokerURL需要改成使用failover。這樣啟動生產者和消費者后，程序就可以在主從直接自動切換（可以嘗試輪流關閉主從）。

 

生產者代碼如下：

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 {
    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
        String brokerURL = "failover://(tcp://localhost:61616,tcp://localhost:61617)";
        connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD, brokerURL);
        try {
            // 構造從工廠得到連接對象
            connection = connectionFactory.createConnection();
            // 啟動
            connection.start();
            // 獲取操作連接
            session = connection.createSession(Boolean.TRUE,
                    Session.AUTO_ACKNOWLEDGE);
            destination = session.createQueue("FirstQueue");
            // 得到消息生成者
            producer = session.createProducer(destination);
                        producer.setDeliveryMode(DeliveryMode.PERSISTENT);
            while (true) {
                sendMessage(session, producer);
                session.commit();// commit后消息才會發出去
                Thread.sleep(1000);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != connection)
                    connection.close();
            } catch (Throwable ignore) {
            }
        }
    }

    static int i = 1;

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

 

消費者代碼如下：

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
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.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;
        String brokerURL = "failover://(tcp://localhost:61616,tcp://localhost:61617)";
//        String brokerURL = "tcp://localhost:61616";
        connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                brokerURL);
        try {
            // 構造從工廠得到連接對象
            connection = connectionFactory.createConnection();
            // 啟動
            connection.start();
            // 獲取操作連接
            session = connection.createSession(Boolean.FALSE,
                    Session.AUTO_ACKNOWLEDGE);
            destination = session.createQueue("FirstQueue");
            consumer = session.createConsumer(destination);
            consumer.setMessageListener(new MyListener());
            System.out.println("started...");
            while(true){
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != connection)
                    connection.close();
            } catch (Throwable ignore) {
            }
        }
    }
}
class MyListener implements MessageListener{
    
    public void onMessage(Message message) {
        TextMessage textMessage = (TextMessage) message;
        try {
            System.out.println("收到消息:"+textMessage.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

 

以上代碼部分摘自網絡

 

這是配置主從的一個方案，還有一種方案是使用文件系統。