ActiveMQ集群下的消息回流功能

------------------------------------------------------------------

"丟失"的消息

　　如果有broker1和broker2通過networkConnector連接，有一個consumer1連接到broker1,一個consumer2連接到broker2，程序往broker1上面發送30條消息，這時consumer2連接到broker2消費消息，當consumer2消費了15條消息時，broker2掛掉了。 但是還剩下15條消息在broker2上面，這些消息就好像消息了，除非broker2重啟了，然后有消費者連接到broker2上來消費消息，遇到這樣的情況該怎么辦呢？

　　從5.6版本起，在destinationPolicy上新增的選擇replayWhenNoConsumers，這個選項使得broker2上有需要轉發的消息但是沒有消費者時，把消息回流到它原來的broker1上，同時需要把enableAudit設置為false,為了防止消息回流后被當做重復消息而不被分發，activemq.xml配置如下：

配置允許雙向連接

先運行發送消息程序，然后運行61716端口的程序接收消息，再接收了5條后將程序斷掉，再去運行61616端口的程序，也是可以接收消息的，這樣就實現了消息的回流。

消息發送程序：

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class JmsSend {
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.81:61616");
        Connection connection = connectionFactory.createConnection();
        connection.start();
        
        Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
        Destination destination = session.createQueue("my-queue");
        
        MessageProducer producer = session.createProducer(destination);
        for(int i = 0;i < 10;i++){
            TextMessage message = session.createTextMessage("message,1212 --->" + i);
            Thread.sleep(1000);
            //通過生產者發出消息
            producer.send(message);
        }
        session.commit();
        session.close();
        connection.close();
    }
}

 61616端口接收消息程序：

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.ActiveMQConnectionFactory;

public class JmsReceiver {
    public static void main(String[] args) throws Exception {
        ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://192.168.1.81:61616");
        Connection connection =  cf.createConnection();
        connection.start();
        
        Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
        Destination destination = session.createQueue("my-queue");
        MessageConsumer consumer = session.createConsumer(destination);
        int i = 0;
        while(i < 10){
            Thread.sleep(3000);
            i++;
            TextMessage message = (TextMessage)consumer.receive();
            session.commit();
            System.out.println("111接收到的消息是:"+message.getText());
        }
        session.close();
        connection.close();
    }
}

 61716端口接收消息程序：

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.ActiveMQConnectionFactory;

public class JmsReceiver2 {
    public static void main(String[] args) throws Exception {
        ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://192.168.1.81:61716");
        Connection connection =  cf.createConnection();
        connection.start();
        
        Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
        Destination destination = session.createQueue("my-queue");
        MessageConsumer consumer = session.createConsumer(destination);
        int i = 0;
        while(i < 10){
            Thread.sleep(3000);
            i++;
            TextMessage message = (TextMessage)consumer.receive();
            session.commit();
            System.out.println("222接收到的消息是:"+message.getText());
        }
        session.close();
        connection.close();
    }
}