ActiveMQ（二十八）--Destination高级特性2

一、Visual Destinations

测试：

先把配置文件整理一下

重启mq

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class PersistenceSender {
    public static void main(String[] args) throws JMSException {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");

        Connection connection = connectionFactory.createConnection();
        Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
        Destination destination = session.createTopic("VirtualTopic.MyTopic3");
        MessageProducer producer = session.createProducer(destination);
        producer.setDeliveryMode(DeliveryMode.PERSISTENT);
        connection.start();

        for (int i = 0; i < 3; i++) {
            TextMessage message = session.createTextMessage("message333--" + i);
            producer.send(message);
        }

        session.commit();
        session.close();
        connection.close();
    }
}

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class QueueReceiver1 {
    public static void main(String[] args) throws JMSException, InterruptedException {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
        for (int i = 0; i < 3; i++) {
            Thread t = new MyThread(connectionFactory);
            t.start();
            Thread.sleep(1000);
        }
    }
}

class MyThread extends Thread {

    private ConnectionFactory connectionFactory = null;

    public MyThread(ConnectionFactory connectionFactory) {
        this.connectionFactory = connectionFactory;
    }

    @Override
    public void run() {
        try {
            final Connection connection = connectionFactory.createConnection();
            connection.start();
            final Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createQueue("Consumer.A.VirtualTopic.MyTopic3");
            MessageConsumer consumer = session.createConsumer(destination);
            consumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    TextMessage txtMsg = (TextMessage) message;
                    try {
                        System.out.println("Receiver111===" + txtMsg.getText());
                        session.commit();
                        session.close();
                        connection.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

先启动消费者。然后，启动生产者。消费者成功收到消息。在3个消费者的角度来看，他们是从queue里面拿消息的。

把消费者2加进来

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class QueueReceiver2 {
    public static void main(String[] args) throws JMSException, InterruptedException {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
        for (int i = 0; i < 3; i++) {
            Thread t = new MyThread2(connectionFactory);
            t.start();
            Thread.sleep(1000);
        }
    }
}

class MyThread2 extends Thread {

    private ConnectionFactory connectionFactory = null;

    public MyThread2(ConnectionFactory connectionFactory) {
        this.connectionFactory = connectionFactory;
    }

    @Override
    public void run() {
        try {
            final Connection connection = connectionFactory.createConnection();
            connection.start();
            final Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createQueue("Consumer.B.VirtualTopic.MyTopic3");
            MessageConsumer consumer = session.createConsumer(destination);
            consumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    TextMessage txtMsg = (TextMessage) message;
                    try {
                        System.out.println("Receiver222===" + txtMsg.getText());
                        session.commit();
                        session.close();
                        connection.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

运行一下消费者2.

再运行生产者，消费者1，消费者2。

消费者1，消费者2各自都能收到3条消息。

测试：

 

二、Mirrored Queues

测试：

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class QueueSender {

    public static void main(String[] args) throws JMSException, InterruptedException {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.128: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("message999--" + i);
            producer.send(message);

        }

        session.commit();
        session.close();
        connection.close();
    }
}

运行生产者

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class PersistenceReceiver {
    public static void main(String[] args) throws JMSException {

        ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
        Connection connection = factory.createConnection();
        connection.setClientID("cc1");
        Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
        Topic topic = session.createTopic("my-queue.qmirror");
        TopicSubscriber consumer = session.createDurableSubscriber(topic, "t1");

        connection.start();

        Message message = consumer.receive();
        while (message != null) {
            TextMessage txtMsg = (TextMessage) message;
            System.out.println("收到消息：" + txtMsg.getText());
            message = consumer.receive(1000L);
        }
        session.commit();
        session.close();
        connection.close();
    }
}

持久化的消费者。先注册一下。然后，再消费。

三、Per Destination Policies

 

https://blog.csdn.net/zhanglh046/article/details/52830624