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

时间：2023-03-09 21:35:55
分布式-信息方式-ActiveMQ的Destination高级特性3

虚拟destination用来创建逻辑destination,客户端可以通过它来生产和消费消息，它会把消息映射到物理destination.

ActiveMQ支持2种方式：

1：虚拟主题（Virtual Topics）

2：组合Destinations（Composite Destinations）

为什么使用虚拟主题？

ActiveMQ只有在持久订阅才是持久化的。持久订阅时，每一个持久订阅者，都相当于一个queue的客户端，它会收取所有消息。这种情况下存在两个问题：

第一：同一应用内消费者端护在均衡的问题。也就是说一个应用程序内的持久化消息，不能使用对个消费者共同承担消息处理能力。因为每个消费者都会获取所有消息。因为每一个消费者都会获取所有信息。

Queue到时可以解决这个问题，但broker端又不能将消息发送到多个应用端，所以纪要发布订阅，又要让消费者分组，这个功能JMS本身是没有的

第二：同一应用内消费者端failover问题，由于只能使用单个的持久订阅者，如果这个订阅者出错，则应用就无法处理，系统的健壮性不高。

如何使用虚拟topic?

第一：对于消息发布者来说，就是一个正常的topic,名称以VirtualTopic.开始，比如VirtualTopic.Orders,代码示例如下：

Topicdestination = session.createTopic("VirtualTopic.Orders");

第二：对于消息接收端来说，是个队列，不同应用里使用不同的前缀作为队列名称，即可表明自己的身份即可实现消费端应用分组。

例如Consumer.A.VirtualTopic.Orders说明它是名称为A的消费端，同理Consumer.B VirtualTopic.Orders说明是一名称为B的消费端。可以在同一个应用中使用多个消费者消费这个队列

又因为不同应用使用的topic名称不一样，前缀不同，所以不同应用中都可以接受到全部消息。每一个客户端相当于一个持久订阅者，而且这个客户端可以使用多个消费者共同来承担消费任务。

代码示例：

Destination dest = session.createQueue("Consumer.A.VirtualTopic.Orders");

代码如下：

package test.mq.visualDestinations;

import javax.jms.Connection;

import javax.jms.ConnectionFactory;

import javax.jms.DeliveryMode;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.MapMessage;

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

           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.createTopic("VirtualTopic.mytopic1");

        MessageProducer messageProducer=session.createProducer(destination);

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

             TextMessage textMessage=session.createTextMessage();

             textMessage.setText("我是TOM ID为"+i);

             messageProducer.send(textMessage);

             System.out.println("生产者："+textMessage.getText());

        }

         session.commit();

         session.close();

         connection.close();

    }

}
package test.mq.visualDestinations;

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("Consumer.A.VirtualTopic.mytopic1");

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

    }

}
package test.mq.visualDestinations;

import java.util.Enumeration;

import javax.jms.Connection;

import javax.jms.ConnectionFactory;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.MessageConsumer;

import javax.jms.MessageListener;

import javax.jms.Session;

import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Receiver2{

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

        ConnectionFactory   connectionFactory=new ActiveMQConnectionFactory(

                "tcp://localhost:61616"

                );

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

            Thread    t=new MyThread2(connectionFactory);

            t.start();

            try {

                Thread.sleep(1000l);

            } catch (InterruptedException e) {

                // TODO Auto-generated catch block

                e.printStackTrace();

            }

        }

    }

}

class MyThread2 extends Thread{

         private ConnectionFactory   connectionFactory=null;

         public  MyThread2(ConnectionFactory   connectionFactory){

         this.connectionFactory = connectionFactory;

         }

       public void run(){

            try {

                final Connection  connection = connectionFactory.createConnection();

                connection.start();

                Enumeration names=connection.getMetaData().getJMSXPropertyNames();

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

                Destination destination=session.createQueue("Consumer.B.VirtualTopic.mytopic1");

                MessageConsumer Consumer=session.createConsumer(destination);

                Consumer.setMessageListener(new MessageListener() {

                    @Override

                    public void onMessage(Message msg) {

                    TextMessage     txtmsg=(TextMessage) msg;

                    try {

                        System.out.println("接收信息2--->"+txtmsg.getText());

                    } catch (JMSException e1) {

                        e1.printStackTrace();

                    }

                    try {

                        session.commit();

                    } catch (JMSException e) {

                        e.printStackTrace();

                    }

                    try {

                        session.close();

                    } catch (JMSException e) {

                        // TODO Auto-generated catch block

                        e.printStackTrace();

                    }

                    try {

                        connection.close();

                    } catch (JMSException e) {

                        // TODO Auto-generated catch block

                        e.printStackTrace();

                    }

                    }

                });

            } catch (JMSException e) {

                // TODO Auto-generated catch block

                e.printStackTrace();

            }

       }

    }

其实把消费者队列化了。

修改虚拟主题的前缀：

默认前缀是VirtualTopic.>

自定义消费虚拟地址默认格式：Consumer.*.VirtualTopic.>

修改配置：

<broker xmlns="http://activemq.apache.org/schema/core">

       <destinationInterceptors>

              <virtualDestinationInterceptor>

                     <virtualDestinations>

                            <virtualTopic name=">" prefix="VirtualTopicConsumers.*." selectorAware="false" />

                     </virtualDestinations>

              </virtualDestinationInterceptor>

       </destinationInterceptors>

</broker>

相关文章