1. 程式人生 > >ActiveMQ 映象佇列Mirrored Queues

ActiveMQ 映象佇列Mirrored Queues

概述
ActiveMQ中每個queue中的訊息只能被一個consumer消費。然而,有時候你可能希望能夠監視生產者和消費者之間的訊息流。你可以通過使用Virtual Destinations 來建立一個virtual queue 來把訊息轉發到多個queues中。但是 為系統中每個queue都進行如此的配置可能會很麻煩。
使用
ActiveMQ支援Mirrored Queues。Broker會把傳送到某個queue的所有訊息轉發到一個名稱類似的topic,因此監控程式只需要訂閱這個mirrored queue topic。為了啟用Mirrored Queues,首先要將BrokerService的useMirroredQueues屬性設定成true,然後可以通過destinationInterceptors設定其它屬性,如mirror topic的字首,預設是“VirtualTopic.Mirror.”。
比如修改後綴的配置示例如下:
<destinationInterceptors>

    <mirroredQueue copyMessage="true" />    <!-- postfix=".qmirror" prefix="" -->

</destinationInterceptors>



當配置完成之後,啟動activemq服務傳送queue時

public class QueueSender {
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new
ActiveMQConnectionFactory("tcp://127.0.0.1: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<3; i++) { TextMessage message = session.createTextMessage("message--"+i); Thread.sleep
(1000); //通過訊息生產者發出訊息 producer.send(message); } session.commit(); session.close(); connection.close(); } }



會生成映象的queue的topic

所有需要接受該訊息的Client只需要註冊該topic,接受訊息即可

public class PersistanceReceiver2 {
    public static void main(String[] args) throws Exception {
        ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
        Connection connection = cf.createConnection();
        connection.setClientID("cc1");//設定ClientID
final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
        Topic destination = session.createTopic("VirtualTopic.Mirror.my-queue");
        //建立持久化使用者
TopicSubscriber consumer = session.createDurableSubscriber(destination,"T1");
        connection.start();
        Message message = consumer.receive();
        while(message!=null) {
            TextMessage txtMsg = (TextMessage)message;
            session.commit();
            System.out.println("收到消 息:" + txtMsg.getText());
            message = consumer.receive(1000L);
        }
        session.close();
        connection.close();
    }
}