ActiveMQ消息中间件之队列模式和主题模式详解

一、ActiveMQ消息中间件
在传统的消息发送和接收模式上，一般是以同步的方式来发送接收消息，以同步的方式来推送消息对我们的服务有时造成了很大的影响，比如当我们的服务器出现了故障，客户端推送消息到服务器，但是服务器此时出现问题它不会返回给我们的客户端它任何接收到消息的信息，这是我们的客户端一直处于等待状态，这样不仅浪费资源而且有时会对我们的产品造成很大的影响，这是我们有一个好的主意，退出了JMS消息服务，作为开源的JMS Provider，ActiveMQ帮助我们解决了这一麻烦的事情，首先我们来看一下使用消息中间件的图解（自己手动画的，有点丑，不要介意哈）。

在这里我们可以看到我们如果采用消息中间件的方式来推送我们的消息，这是当我们的客户端想给服务器发送消息的时候，我们只要将消息发送出去，经过消息中间件，至于消息中间件什么时候推送消息给服务器，这些客户端并不关系，它直接返回，因为采用的是异步的方式，即使现在服务器我们的服务器忙碌或者出现问题，也不影响我们消息的推送，当下次服务器启动的时候自然可以接收到消息。这里举个简单的例子就会对消息中间件有一个更好的理解，假如你要向你的老板请假，但此时你的boss正在开会，而你又急着赶飞机，这是该怎么办呢？当然你可以给你的boss打电话，但是也许你一个电话导致会议中断，这时造成了不必要的麻烦，所以打电话似乎不太好，但是如果要你等的话，飞机不等你，这是我们想到一个更好的办法，我们可以给boss留言，这样你就可以去赶你的飞机，而老板开完会自然会看到你的留言，这样都不耽误，所以，这样就轻松解决了我们的问题。采用消息中间件的平台比如微信等等。

这里我们主要介绍ActiveMQ消息中间件，它有两种消息模式，一种是队列模式，一种是主题模式。下面我们分别来介绍。

二、队列模式
什么是队列模式呢？所谓队列模式，就是我们的消息消息生产者（Producer）向一个目的地(Desitinate)存放我们的消息，如果是多个消费者我们可以它们会分别消费这些消息，没个消息只能供一个消费者使用，下面我们用图解来清除表示：

从上面的图示我们可以很清除的看出队列模式的含义，即每一个消息只能被一个消费者消费，当有一个消费者使用该消息后，即从我们的队列中取出该消息。

三、主题模式
如果我们上面理解了队列模式，这里我们很容易理解主题模式，与之相反的，每个消息可以供多个消费者使用，如图所示。

四、ActiveMQ实战
经过上面的一番讲解，相信大家都对两种模式有了一个清楚的概念，下面，我们来看一个ActiveMQ的工作流程图。

这里使用一个工厂类来创建一个连接，通过连接来创建一个会话，然后通过会话来创建消费者和生产者，这里有一个消息监听器用来监听消息队列的，生产者和消费者通过这个监听器来手法消息，先这样简单了解一下，我们通过例子来详细了解。这里引用一句话，学习最好的方式就是实战（这是spring3.x这本书中的一句话，原话记不太清了，大概就是这样，哈哈）。下面我们来创建项目编写我们的代码。

1.首先，我们创建一个maven工程，下面是所需要的依赖。

<dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-core</artifactId>
        <version>5.7.0</version>
    </dependency>

首先，我们创建一个队列模式下的生产者和消费者

Producer

public class AppProducter {
    private static final String URL="tcp://127.0.0.1:61616";
    private static final String QueueName="QueueTest";
    public static void main(String[] args) throws JMSException {
        //创建工厂类
        ConnectionFactory connectionFactory=new ActiveMQConnectionFactory(URL);
        //创建连接
        Connection conn=connectionFactory.createConnection();
        //启动连接
        conn.start();
        //创建会话
        Session session=conn.createSession(false,Session.AUTO_ACKNOWLEDGE);
        //创建目的地(消息发送的目标)
        Destination destination=session.createQueue(QueueName);
        //创建消息发布者
        MessageProducer producer= session.createProducer(destination);
        for(int i=0;i<100;i++) {
            //创建消息
            TextMessage message=session.createTextMessage("test"+i);
            producer.send(message);
            System.out.println("发送消息"+i+" "+message.getText());
        }
        //关闭连接
            conn.close();
    }
}

Consumer

package cn.shinelon.queue;

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 AppConsumer {
    private static final String URL="tcp://127.0.0.1:61616";
    private static final String QueueName="QueueTest";
        public static void main(String[] args) throws JMSException {
            //创建工厂类
            ConnectionFactory connectionFactory=new ActiveMQConnectionFactory(URL);
            //创建连接
            Connection conn=connectionFactory.createConnection();
            //启动连接
            conn.start();
            //创建会话
            Session session=conn.createSession(false,Session.AUTO_ACKNOWLEDGE);
            //创建目的地(消息发送的目标)
            Destination destination=session.createQueue(QueueName);
            //创建消息发送者
            MessageConsumer consumer=session.createConsumer(destination);
            //创建监听器 
            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message msg) {
                    TextMessage message=(TextMessage) msg;
                    try {
                        System.out.println("接收消息："+" "+message.getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
            //因为接收消息是异步的，所以不能关闭连接
    }
}

上面代码就不用多加解释了，可以看代码注释，运行上面两个程序，我们会发现一个消息只能被一个消费者使用（自己动手试试）

接着，我们创建主题模式下的生产者和消费者的运行代码，基本和上面一样，这里贴出完成的代码。

Producer

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;

public class AppProducter {
    private static final String URL="tcp://127.0.0.1:61616";
    private static final String topicName="topicTest";
    public static void main(String[] args) throws JMSException {
        //创建工厂类
        ConnectionFactory connectionFactory=new ActiveMQConnectionFactory(URL);
        //创建连接
        Connection conn=connectionFactory.createConnection();
        //启动连接
        conn.start();
        //创建会话
        Session session=conn.createSession(false,Session.AUTO_ACKNOWLEDGE);
        //创建目的地(消息发送的目标)
        Destination destination=session.createTopic(topicName);
        //创建消息发布者
        MessageProducer producer= session.createProducer(destination);
        for(int i=0;i<100;i++) {
            //创建消息
            TextMessage message=session.createTextMessage("test"+i);
            producer.send(message);
            System.out.println("发送消息"+i+" "+message.getText());
        }
        //关闭连接
            conn.close();
    }
}

Consumer

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;
/**
 * 与队列模式不同的是主题模式要提前订阅消息（即先启动消费者），不然消费者不会受到任何消息
 * @author Shinelon
 *
 */
public class AppConsumer {
    private static final String URL="tcp://127.0.0.1:61616";
    private static final String topicName="topicTest";
        public static void main(String[] args) throws JMSException {
            //创建工厂类
            ConnectionFactory connectionFactory=new ActiveMQConnectionFactory(URL);
            //创建连接
            Connection conn=connectionFactory.createConnection();
            //启动连接
            conn.start();
            //创建会话
            Session session=conn.createSession(false,Session.AUTO_ACKNOWLEDGE);
            //创建目的地(消息发送的目标)
            Destination destination=session.createTopic(topicName);
            //创建消息发送者
            MessageConsumer consumer=session.createConsumer(destination);
            //创建监听器 
            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message msg) {
                    TextMessage message=(TextMessage) msg;
                    try {
                        System.out.println("接收消息："+" "+message.getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
            //因为接收消息是异步的，所以不能关闭连接
    }
}

这时我们运行代码会发送每一个消息可以被多个消费者使用，当一个消息被一个消费者使用户，它不会取出来，而是等待被下一个消费者使用（与队列模式恰好相反）。

通过上面的介绍，我们了解到了ActiveMQ在两种不同模式下实现简单消息的推送，在后面的文章中我们会实现与spring的整合，欢迎讨论。

原文链接：https://blog.csdn.net/qq_37142346/article/details/78756708

ActiveMQ消息中间件之队列模式和主题模式详解

热门标签