分布式-信息方式-ActiveMQ的Destination高级特性1

ActiveMQ的Destination高级特性

Destination高级特性----->Composite Destinations
　　组合队列Composite Destinations ： 允许用一个虚拟的destination代表多个destinations,这样就可以通过composite destinations在一个操作中同时向多个queue/topic发送消息。
　　有两种实现方式：
　　　　第一种:在客户端编码实现
　　　　第二种：在activemq.xml配置文件中实现

　　第一种：在客户端编码实现
　　　　在composite destinations中，多个destination之间采用","分隔。如下：这里有2个destination  "my-queue1"和"my-queue2"

信息生产者

package test.mq.destination;

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 Sender {

    public static void main(String[] args) throws Exception {

        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");

        Connection connection = connectionFactory.createConnection();

        connection.start();

     Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);

     Destination destination = session.createQueue("my-queue1,my-queue2");

        MessageProducer producer = session.createProducer(destination);

        for(int i = 0;i < 5;i++){

            TextMessage message = session.createTextMessage("messageAAA --->" + i);

            //通过生产者发出消息

            producer.send(message);

        }

        session.commit();

        session.close();

        connection.close();

    }

}

运行消息发送程序后，去访问http://localhost:8161/admin/queues.jsp，可以看到如下：

消费者1

package test.mq.destination;

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 Receiver1 {

    public static void main(String[] args) throws Exception {

        ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://localhost:61616");

        Connection connection =  cf.createConnection();

        connection.start();

    Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);

    Destination destination = session.createQueue("my-queue1");

        MessageConsumer consumer = session.createConsumer(destination);

        int i = 0;

        while(i < 5){

            Thread.sleep(1000);

            i++;

            TextMessage message = (TextMessage)consumer.receive();

            session.commit();

            System.out.println("1接收到的消息是:"+message.getText());

        }

        session.close();

        connection.close();

    }

}

消费者2

package test.mq.destination;

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 Receiver2 {

    public static void main(String[] args) throws Exception {

        ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://localhost:61616");

        Connection connection =  cf.createConnection();

        connection.start();

    Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);

    Destination destination = session.createQueue("my-queue2");

        MessageConsumer consumer = session.createConsumer(destination);

        int i = 0;

        while(i < 5){

            Thread.sleep(1000);

            i++;

            TextMessage message = (TextMessage)consumer.receive();

            session.commit();

            System.out.println("2接收到的消息是:"+message.getText());

        }

        session.close();

        connection.close();

    }

}

第二种：在activemq.xml配置文件中实现

<destinationInterceptors>

           <virtualDestinationInterceptor>

              <virtualDestinations>

                     <compositeQueue name="MY.QUEUE">

                            <forwardTo>

                                   <queue physicalName="my-queue1" />

                                   <queue physicalName="my-queue2" />

                            </forwardTo>

                     </compositeQueue>

              </virtualDestinations>

           </virtualDestinationInterceptor>

        </destinationInterceptors>

生产者：

package test.mq.destination;

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 Sender {

    public static void main(String[] args) throws Exception {

        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");

        Connection connection = connectionFactory.createConnection();

        connection.start();

     Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);

    // Destination destination = session.createQueue("my-queue1,my-queue2");

     Destination destination = session.createQueue("MY.QUEUE");

        MessageProducer producer = session.createProducer(destination);

        for(int i = 0;i < 5;i++){

            TextMessage message = session.createTextMessage("messageAAA --->" + i);

            //通过生产者发出消息

            producer.send(message);

        }

        session.commit();

        session.close();

        connection.close();

    }

}

消费者1和2  与上面的代码一样，这里就不重复了。