activeMQ中的Virtual Topics详解及使用

2017-04-14  本文已影响1283人  吴世浩

一、好言

太远容易生疏，太近容易情尽。

二、背景

最近接手项目，公司的MQ做了一层封装，挺好用的，会有一片文章记载，然后在其中我们使用了<a href="http://activemq.apache.org/virtual-destinations.html">虚拟话题</a>的概念,这个我没有使用过，之前一直都是使用单纯的队列或者topic，所以就查询资料，自己配制写测试案列看看实际效果。

三、直接先上测试代码

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import javax.jms.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by Mahone Wu on 2017/4/12.
 */
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration("/applicationContext-v.xml")
public class TopicTest {

    private Logger logger = LoggerFactory.getLogger(SpringJmsTopicTest.class);

    ActiveMQConnectionFactory factoryA;

    Session session;


    @Before
    public void init(){
        try{
            factoryA = getAMQConnectionFactory();
            ActiveMQConnection conn = (ActiveMQConnection) factoryA.createConnection();
            conn.start();
            session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        }catch (Exception e){
            e.printStackTrace();
        }

    }

    @Test
    public void testNormalTopic(){
        try {
            ActiveMQTopic queue = new ActiveMQTopic(getNormalTopicName());
            MessageConsumer consumer1 = session.createConsumer(queue);
            MessageConsumer consumer2 = session.createConsumer(queue);
            final AtomicInteger count = new AtomicInteger(0);
            MessageListener listenerA = new MessageListener() {
                public void onMessage(javax.jms.Message message) {

                    try{
                        int index =count.incrementAndGet();
                        logger.info("index={}---------->receive from {},消息={}",index,getNormalTopicName(),((TextMessage)message).getText());
                        Thread.sleep(10L);
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
            };
            consumer1.setMessageListener(listenerA);
            consumer2.setMessageListener(listenerA);

            MessageProducer producer = session.createProducer(new ActiveMQTopic(getNormalTopicName()));
            int index = 0;
            while (index++ < 10) {
             //   logger.info("{},{}",index < 100,index + " message.");
                TextMessage message = session.createTextMessage(index
                        + " message.");
                producer.send(message);
                Thread.sleep(5L);
            }
        }catch (Exception e){
            e.printStackTrace();
        }

        try {
            System.in.read();
        }catch (Exception e){
            e.printStackTrace();
        }

    }



    @Test
    public void testNormalVirtualTopic(){
        try{
            Queue queue = new ActiveMQQueue(getVirtualTopicConsumerName());
            MessageConsumer consumer1 = session.createConsumer(queue);
            MessageConsumer consumer2 = session.createConsumer(queue);
            final AtomicInteger count = new AtomicInteger(0);

            MessageListener listenerA = new MessageListener() {
                public void onMessage(javax.jms.Message message) {
                    try {
                        int index = count.getAndIncrement();
                        logger.info("index={}---------->receive from {},消息={}", index, getNormalTopicName(), ((TextMessage) message).getText());
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
            };
            consumer1.setMessageListener(listenerA);
            consumer2.setMessageListener(listenerA);
            MessageProducer producer = session.createProducer(new ActiveMQTopic(getNormalVTopicName()));
            int index = 0;
            while (index++ < 10) {
                TextMessage message = session.createTextMessage(index
                        + " message.");
                producer.send(message);
            }
        }catch (Exception e){
            e.printStackTrace();
        }

    }

    @Test
    public void testVirtualTopic(){
        try {
            Queue queue = new ActiveMQQueue(getVirtualTopicConsumerNameA());
            MessageConsumer consumer1 = session.createConsumer(queue);
            MessageConsumer consumer2 = session.createConsumer(queue);
            MessageConsumer consumer3 = session.createConsumer(new ActiveMQQueue(getVirtualTopicConsumerNameB()));

            final AtomicInteger countA = new AtomicInteger(0);
            MessageListener listenerA = new MessageListener() {
                public void onMessage(javax.jms.Message message) {
                    try {
                        int index = countA.getAndIncrement();
                        logger.info("A index={}---------->receive from {},消息={}", index, getNormalTopicName(), ((TextMessage) message).getText());
                    }catch (Exception e){
                        logger.error(""+e);
                        e.printStackTrace();
                    }
                }
            };
            consumer1.setMessageListener(listenerA);
            consumer2.setMessageListener(listenerA);
            final AtomicInteger countB = new AtomicInteger(0);
            MessageListener listenerB = new MessageListener() {
                public void onMessage(javax.jms.Message message) {
                    try {
                        int index = countB.getAndIncrement();
                        logger.info("B index={}---------->receive from {},消息={}", index, getNormalTopicName(), ((TextMessage) message).getText());
                    }catch (Exception e){
                        e.printStackTrace();
                        logger.error(""+e);
                    }
                }
            };
            consumer3.setMessageListener(listenerB);
            MessageProducer producer = session.createProducer(new ActiveMQTopic(getVirtualTopicName()));
            int index = 0;
            while (index++ < 10) {
                TextMessage message = session.createTextMessage(index
                        + " message.");
                producer.send(message);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    private ActiveMQConnectionFactory getAMQConnectionFactory(){
        return new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
    }

    private static String getNormalTopicName(){
        return "normal.TEST";
    }

    private static String getNormalVTopicName(){
        return "VirtualTopic.NORMAL";
    }

    private static String getVirtualTopicName(){
        return "VirtualTopic.TEST";
    }

    private static String getVirtualTopicConsumerName(){
        return "Consumer.normal.VirtualTopic.NORMAL";
    }

    private static String getVirtualTopicConsumerNameA(){
        return "Consumer.A.VirtualTopic.TEST";
    }

    private static String getVirtualTopicConsumerNameB(){
        return "Consumer.B.VirtualTopic.TEST";
    }
}

applicationContext-v.xml：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns:p="http://www.springframework.org/schema/p" xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
    xsi:schemaLocation="
            http://www.springframework.org/schema/beans
            http://www.springframework.org/schema/beans/spring-beans.xsd
            http://www.springframework.org/schema/context
            http://www.springframework.org/schema/context/spring-context.xsd">
</beans>

testNormalTopic打印结果：

testNormalTopic.png
结论：
从这个里面我们可以看出，consumer1,consumer2都监听了normal.TEST,所以结果是打印了20条数据

testNormalVirtualTopic打印结果：

testNormalVirtualTopic.png
结论：
虚拟的方式，consumer1，consumer2，都订阅了VirtualTopic.NORMAL，所以按照配制消费端Consumer.normal.VirtualTopic.NORMAL的方式，结果只打印了10条数据，所以此时consumer1，consumer2只有一个接收成功。

testVirtualTopic打印结果


testVirtualTopic.png

结论：
这个跟上面这个类似，对比更佳明显的说明了VirtualTopic中，A,B是两个不同的应用，所以打印了20条，如果是同一个应用，则只会又一个接收成功。

Consumer.A.VirtualTopic.Test
1、Consumer ,VirtualTopic 为系统配置，勿做修改。**
2、A为消费方的系统名称。
3、Test 为根据业务定义的消息地址
<a href="http://activemq.apache.org/virtual-destinations.html">官方文档</a>有做相关说明。

四、虚拟主题用处

摘自网上：

  1. 同一应用内consumer端负载均衡的问题：同一个应用上的一个持久订阅不能使用多个consumer来共同承担消息处理功能。因为每个都会获取所有消息。queue模式可以解决这个问题，broker端又不能将消息发送到多个应用端。所以，既要发布订阅，又要让消费者分组，这个功能jms规范本身是没有的。
  1. 同一应用内consumer端failover的问题：由于只能使用单个的持久订阅者，如果这个订阅者出错，则应用就无法处理消息了，系统的健壮性不高。
    对于上述的表述个人觉得理解起来好纠结，因为这里又涉及到持久化问题，对于持久订阅的意义可以看这篇<a href="http://blog.csdn.net/wenlixing110/article/details/53032324">文章</a>

所以我个人觉得这里解决的就是对于如果单纯的使用topic方式，那么如果消费端部署的是集群方式，那么每一个都订阅了，在发送消息的时候，集群中的每一个订阅者都有可能收到，那么这不是我们想要的效果；可能上面说的有这么一个方面还有一个就是涉及到持久定于的健壮性问题。
所以virtualtopic我们可以理解为是queue和topic的结合，即是，使用topic的一对多的广播功能，又需要在集群的时候，只有一个收到，也就是队列的一对一的特效。

五：代码

代码放github了，主要在单元测试里面可以看看
地址：https://github.com/MahoneWu/mq

上一篇下一篇

猜你喜欢

热点阅读