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();
    }
}