ActiveMQ入门教程(三)-------java编码ActiveMQ实现(队列QUEUE、主题TOPIC)

我不是女神ヾ 2021-09-01 08:31 405阅读 0赞

# java编码ActiveMQ实现(队列QUEUE、主题TOPIC) #

## 前言 ##

在前一章 [ActiveMQ入门教程(二)-------java实现MQ标准API及JMS整体架构](https://blog.csdn.net/qq_43419105/article/details/103709585)中我们创建了一个IDEA的maven项目，这一节就以上一节的Maven项目开始，用java代码实现消息中间件MQ的消息发送以及接收。

## 在点对点的消息传递域中，目的地被称之为队列(Queue) ##

### 点对点消息传递域作用如下 ###

1.  每个消息只能有一个消费者，类似1对1的关系。好比个人快递自己领取自己的。
2.  消息的生产者和消费者之间没有时间上的关联性。无论消费者在生产者发送消息的时候是否处于运行状态，消费者都可以提取消息。好比我们的发送消息，发送者发送之后不见得接受者会即收即看。
3.  消息被消费后队列中不会在存储，所以消费者不会消费到已经被消费掉的消息。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70]

### 生产者生产消息 ###

public class JmsProduce {
    
        public static  final  String ACTIVEMQ_URL="tcp://127.0.0.1:61616";
        public  static  final  String QUEUE_NAME="queue01";
        public static void main(String[] args) throws JMSException {
            //1.创建连接工场
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            //2.通过连接工场获得连接connection,并启动
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //3. 创建会话Session
            //两个参数第一个参数表示事物，第二个参数表示签收AUTO_ACKNOWLEDGE表示默认
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地（分为两种一个是队列Queue，一个是主题Topic）
            Queue queue = session.createQueue(QUEUE_NAME);
            //5.创建消息的生产者
            MessageProducer messageProducer = session.createProducer(queue);
            //6.通过使用messageProducer生产3条消息发送到MQ的队列里面
            for (int i=1;i<=3 ; i++){
                //7.创建消息
                TextMessage textMessage = session.createTextMessage("msg------"+i);//理解为一个字符串
                //8.通过messageProducer发送给mq
                messageProducer.send(textMessage);
            }
            //9.关闭资源
            messageProducer.close();
            session.close();
            connection.close();
            //10.请求成功时发送
            System.out.println("******消息发送到MQ成功");
        }
    }

**成功标志**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][20191226143810606.jpg]  
**字段说明**  
![在这里插入图片描述][20191226143922940.jpg]  
**生产者总结**  
当有一个消息进入这个队列时，等待消费的消息是1，进入队列的消息是1；  
当消息消费后，等待消费的消息是0，进入队列的消息是1，出队列的消息是1；  
再来一条消息时，等待消费的消息是1，进入队列的消息就是2；

### 消费者消费消息 ###

/**
     * 消息的消费者
     */
    public class JmsConsumer {
    
        public static  final  String ACTIVEMQ_URL="tcp://127.0.0.1:61616";
        public  static  final  String QUEUE_NAME="queue01";
        public static void main(String[] args) throws JMSException {
    
            //1.创建连接工场
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            //2.通过连接工场获得连接connection,并启动
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //3. 创建会话Session
            //两个参数第一个参数表示事物，第二个参数表示签收AUTO_ACKNOWLEDGE表示默认
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地（分为两种一个是队列Queue，一个是主题Topic）
            Queue queue = session.createQueue(QUEUE_NAME);
            //5.创建消费者
            MessageConsumer messageConsumer = session.createConsumer(queue);
    //同步阻塞方法(receive())；订阅者或者接受者调用MessageConsumer的receive()方法来接受消息，receive方法在能//够接受到消息之前（或超时之前）将一直阻塞
            while(true){
                //接受消息与之前的发送消息的格式对应
                TextMessage textMessage = (TextMessage) messageConsumer.receive();
    //第二种方法带时间等待表示4秒钟之后就会停止消息的消费：
    //TextMessage textMessage = (TextMessage) messageConsumer.receive(4000l);
                if (null != textMessage){
                    System.out.println("****消费者接受消息"+ textMessage.getText());
                }else {
                    break;
                }
            }
            messageConsumer.close();
            session.close();
        }
    }

**成功标志**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 3]

### 消费者消费信息的第二种方式 ###

public class JmsConsumer {
    
        public static  final  String ACTIVEMQ_URL="tcp://127.0.0.1:61616";
        public  static  final  String QUEUE_NAME="queue01";
    
        public static void main(String[] args) throws JMSException, IOException {
            //1.创建连接工场
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            //2.通过连接工场获得连接connection,并启动
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //3. 创建会话Session
            //两个参数第一个参数表示事物，第二个参数表示签收AUTO_ACKNOWLEDGE表示默认
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地（分为两种一个是队列Queue，一个是主题Topic）
            Queue queue = session.createQueue(QUEUE_NAME);
            //5.创建消费者
            MessageConsumer messageConsumer = session.createConsumer(queue);
            //通过监听的方式来消费消息 MessageConsumer messageConsumer1 = session.createConsumer(queue);
            //MessageListener是一个接口 实现接口重写其方法
    /**
     * 异步非阻塞方式（监听器onMessage()）
     * 订阅者或者接受者通过MessageConsumer的setMessageListener(MeaasgeListener listener)注册一个消息监听器
     * 当消息到达之后，系统自动调用监听器MessageListener的onMessage(Message message)方法
     */
    
            messageConsumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    //消费的消息要和接受的消息的格式一致
                    if(null != message && message instanceof  TextMessage){
                        TextMessage textMessage = (TextMessage) message;
                        try {
                            System.out.println("***消费者接受到消息：" + textMessage.getText());
                        } catch (JMSException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });
            System.in.read();//保证控制台不灭
            messageConsumer.close();
            session.close();
            connection.close();
        }
    }

**消费者3大消费情况**  
1.先生产， 只启动一个消费者。问题：1号消费者能消费吗？ yes

2.先生产， 先启动1号消费者，再启动2号消费者。问题：2号消费者还能消费吗？ 启动1号消费者之后都被1号消费者消费完了，所以即使2号消费者挤进来也不会有消费记录了即：1号消费者可以消费，2号消费者没有消费

3.先启动2个消费者，在启动6条消息 2个消费者都会消费消息并且一人一半两个消费者原地待命  
**结果展示**  
![在这里插入图片描述][20191226161857563.jpg]  
![在这里插入图片描述][20191226161911219.jpg]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 4]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 5]  
**消费者两种消费方式总结**

*  同步阻塞方式(receive()):订阅者或者接受者调用MessageConsumer的receive()方法来接收消息，receive方法能在接收到消息之前（或超时之前将一直阻塞）revive方法有两种一种是不传参的一种是传参的，参数及时间
 *  异步非阻塞方式(监听器onMessage())：订阅者或者接受者通过MessageConsumer的setMeaasgeListener(MessageListener listener)注册一个监听器，当消息到达之后，系统自动调用监听器MessageListener的onMeaasge(Message message)方法。

##### 队列案例总结 #####

JMS开发步骤  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 6]

1.  创建一个 connection factory
2.  通过connection factory来创建JMS connection
3.  启动JMS connection
4.  通过connection创建JMS session
5.  创建JMS distination
6.  创建JMS producer或者创建JMS message 并设置destination
7.  创建JMS consumer或者注册一个 JMS message listener
8.  发送或者接受JMS message(s)
9.  关闭所有的JMS资源(connection,session,producer,consumer等)

##### 回顾JDBC操作数据库的通用步骤 #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 7]

## 在发布订阅消息传递域中，目的地被称之为主题（Topic） ##

### 发布/订阅消息传递域的特点 ###

1.  生产者将消息发送到Topic中，每个消息可以有多个消费者，属于1：N的关系
2.  生产者和消费者之间有时间上的相关性。订阅某一个主题的消费者只能消费自它订阅之后发布的消息
3.  生产者生产时，topic不保存消息它是无状态的不落地，假如无人订阅就去生产，那就是一条废消息，所以一般先启动消费者再启动生产者。

JMS规范允许客户创建持久订阅，这在一定程度上放松了时间上的相关性要求。持久订阅允许消费者消费它在未处于激活状态时发送的消息。一句话，好比我们的微信公众号的订阅

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 8]

### 发布主题的生产者 ###

public class JmsTopicProduce {
    
        public static  final  String ACTIVEMQ_URL="tcp://127.0.0.1:61616";
        public  static  final  String TOPIC_NAME="topic01";
    
        public static void main(String[] args) throws JMSException {
            //1.创建连接工场，按照给定的URL地址，采用默认用户名和密码
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            //2.通过连接工场，获得连接的connection 抛出JMSException异常
            Connection connection = activeMQConnectionFactory.createConnection();
            //3.启动JMS connection
            connection.start();
            //4.创建Session；session中有两个参数 第一个叫事物、第二个叫签收
            Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
            //5.创建目的地，（队列是queue/队列是Topic）
            Topic topic = session.createTopic(TOPIC_NAME);
            //6.创建消息的生产者
            MessageProducer messageProducer = session.createProducer(topic);
            //7.通过使用messageProducer生产3条消息发送到MQ的队列里面
            for (int i = 0; i < 3; i++) {
                //8、创建消息
                TextMessage textMessage = session.createTextMessage("*******这是主题消息"+i);//理解为一个字符串
                //9.通过messageProducer发送给mq
                messageProducer.send(textMessage);
            }
            //10.关闭资源
            messageProducer.close();
            session.close();
            connection.close();
            System.out.println("*****消息发送到mq完成");
        }
    }

**成功标志**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 9]

### 订阅主题的消费者 ###

public class JmsTopicConsumer {
    
        public static  final  String ACTIVEMQ_URL="tcp://127.0.0.1:61616";
        public  static  final  String TOPIC_NAME="topic01";
    
        public static void main(String[] args) throws JMSException, IOException {
            //1.创建连接工场
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            //2.通过连接工场获取connection并启动
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //3.创建会话的Session 里面有两个参数 第一个参数是表示事物、第二个参数表示签收
            Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地（主题是Topic,队列是Queue）
            Topic topic = session.createTopic(TOPIC_NAME);
            //5.创建消费者
            MessageConsumer messageConsumer = session.createConsumer(topic);
            //6.启动监听器，通过监听器的方式来获取消息生产者
            /**
             * MessageListener是一个接口，其内部就只有一个onMessage方法，若要实现接口就必须重写它里面的方法
             */
         /*   //第一种方法匿名内部类的写法
            messageConsumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    TextMessage textMessage = (TextMessage) message;
                    try {
                        System.out.println("*********Topic主题接受消息" + textMessage.getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });*/
    
            //第二种写法 jdk8新特性 lambda表达式的写法
           messageConsumer.setMessageListener((message) -> {
               TextMessage textMessage = (TextMessage) message;
               try {
                  System.out.println("*********Topic主题接受消息" + textMessage.getText());
               } catch (JMSException e) {
                  e.printStackTrace();
               }
           });
            //抛出IOException
            System.in.read();
            messageConsumer.close();
            session.close();
            connection.close();
        }
    }

**成功标志**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 10]

### 先启动订阅再启动生产，不然发送的消息就是费消息 ###

![在这里插入图片描述][20191226164004519.jpg]  
**启动3次，出现三个消费者，九条消息**  
![在这里插入图片描述][20191226164102272.jpg]  
实际上只有3条消息被消费

## Topic主题和Queue队列对比总结 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 11]

#### 两大模式比较 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 12]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70]: /images/20210901/8407cc7c61f540308c2ab32b7bbded45.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 1]: /images/20210901/616bef03869341d5a17da4d62d58c80e.png
[20191226143810606.jpg]: /images/20210901/a907b1f4d46c44bf88234a300bf5f117.png
[20191226143922940.jpg]: /images/20210901/d46d3f2bc45f4a098dc4b2ccd971eded.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 2]: /images/20210901/ae8b3733e4d74f838eff6c6ad33d7bbb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 3]: /images/20210901/73f3569e726541e589f0f5d89ff62305.png
[20191226161857563.jpg]: /images/20210901/5982117189f14da3ac51aa1808db5ff5.png
[20191226161911219.jpg]: /images/20210901/e1b3333b6e0543ccbf8366baa8bd664b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 4]: /images/20210901/90f534d7dd674c49a2e92345bf68bed9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 5]: /images/20210901/6ccab184dfe04b359826f2e21d2b33a5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 6]: /images/20210901/9ec83e0c993741279c8b35018a16f329.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 7]: /images/20210901/0a749f4b794143419da80452d261cbe3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 8]: /images/20210901/b9f1bec0458b4b7289bf1ece660f4d0c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 9]: /images/20210901/aa9841ce27e743a2b97f530058bbb66b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 10]: /images/20210901/286ff204cea44d5bb69677ac15320377.png
[20191226164004519.jpg]: /images/20210901/f52d859baf454b75ac0df2b9772937aa.png
[20191226164102272.jpg]: /images/20210901/87eb4ed528b349ba829da5d38f6b73f5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 11]: /images/20210901/85d09f36a3014769961ae95d4211b0ae.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 12]: /images/20210901/79c744d19a1246c3852a1856f16c58fe.png