ActiveMQ--发送消息到队列、从队列接收消息

清疚 2021-08-13 11:59 801阅读 0赞

### JMS编码总体架构 ###

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

### Destincation ###

*  在点对点的消息传递域中，目的地被称为队列（queue）。
 *  在发布/订阅消息传递域中，目的地被称为朱琪（topic）。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]

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

### 案例 ###

发送方编码

package pres.zhang.activemq;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    public class JmsProduce { 
    
        public static final String ACTIVEMQ_URL = "tcp://localhost:61616";
        public static final String QUEUE_NAME = "queue01";
    
        public static void main(String[] args) throws JMSException { 
            //1.创建连接工厂,按照给定的url地址，采用默认用户名和密码
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            //2.通过连接工厂，获得连接connection，并启动访问
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //3.创建回话session
            //两个参数 1.事务 2.签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地（具体是队列还是主题topic）
            Queue queue = session.createQueue(QUEUE_NAME);
            //创建消息的生产者
            MessageProducer messageProducer = session.createProducer(queue);
            //通过使用messageProducer生产3条消息发送到MQ的队列里面
            for (int i = 0; i < 3; i++) { 
                //7.创建消息
                TextMessage textMessage = session.createTextMessage("msg---" + i);//可以理解为字符串
                //8.通过messageProducer发送给MQ
                messageProducer.send(textMessage);
            }
            //9.关闭资源
            messageProducer.close();
            session.close();
            connection.close();;
            System.out.println("消息发布到MQ完成！");
        }
    }

运行代码，然后访问localhost:8161：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]

控制台的说明：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]

接收方消息编码  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]

package pres.zhang.activemq;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    public class JmsConsumer { 
    
        public static final String ACTIVEMQ_URL = "tcp://localhost:61616";
        public static final String QUEUE_NAME = "queue01";
    
        public static void main(String[] args) throws JMSException { 
            //1.创建连接工厂,按照给定的url地址，采用默认用户名和密码
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            //2.通过连接工厂，获得连接connection，并启动访问
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //3.创建回话session
            //两个参数 1.事务 2.签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地（具体是队列还是主题topic）
            Queue queue = session.createQueue(QUEUE_NAME);
            //5.创建消费者
            MessageConsumer messageConsumer = session.createConsumer(queue);
            while(true){ 
                TextMessage textMessage = (TextMessage) messageConsumer.receive();
                if(textMessage != null){ 
                    System.out.println("*****消费者接收到消息：" + textMessage.getText());
                }else{ 
                    break;
                }
            }
            messageConsumer.close();
            session.close();
            connection.close();
        }
    }

运行程序，输出：

*****消费者接收到消息：msg---0
    *****消费者接收到消息：msg---1
    *****消费者接收到消息：msg---2
    *****消费者接收到消息：msg---0
    *****消费者接收到消息：msg---1
    *****消费者接收到消息：msg---2

控制台：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 7]

第一中接收方式：receive()方法

*  无参：阻塞式，如果没有收到消息就一直等待。
 *  long类型参数（毫秒）：超时等待，等待指定的时间仍然没有收到消息则自动释放。

第二种接收方式：监听

package pres.zhang.activemq;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    import javax.xml.soap.Text;
    import java.io.IOException;
    
    public class JmsConsumer { 
    
        public static final String ACTIVEMQ_URL = "tcp://localhost:61616";
        public static final String QUEUE_NAME = "queue01";
    
        public static void main(String[] args) throws JMSException, IOException { 
            //1.创建连接工厂,按照给定的url地址，采用默认用户名和密码
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            //2.通过连接工厂，获得连接connection，并启动访问
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //3.创建回话session
            //两个参数 1.事务 2.签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地（具体是队列还是主题topic）
            Queue queue = session.createQueue(QUEUE_NAME);
            //5.创建消费者
            MessageConsumer messageConsumer = session.createConsumer(queue);
    
            //第一种方式：receive（）
            // while(true){ 
            // TextMessage textMessage = (TextMessage) messageConsumer.receive();
            // if(textMessage != null){ 
            // System.out.println("*****消费者接收到消息：" + textMessage.getText());
            // }else{ 
            // break;
            // }
            // }
            // messageConsumer.close();
            // session.close();
            // connection.close();
    
    
    
            //第二中方式：监听
            messageConsumer.setMessageListener(new MessageListener() { 
                @Override
                public void onMessage(Message message) { 
                    if(message != null && 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();
        }
    }

运行输出：

监听器接收消息：msg---0
    监听器接收消息：msg---1
    监听器接收消息：msg---2

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

### 消费者3大消费情况 ###

1.  先生产： 只启动1号消费者。问题：1号消费者能消费吗？  
    可以
2.  先生产：先启动1号消费者再启动2号消费者。问题：2号消费者还能消费消息吗？  
    1号可以，2号没有，都被一号消费了。
3.  先启动两个消费者，再生产6条消息。问题，消费情况如何？  
    1号消费者：第1，3，5条消息。2号消费者：第2，4，6条。类似轮询。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]: /images/20210813/e7c915e27d2c44bf8f6edac6b5905fc1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]: /images/20210813/5fe8115272f44bd4ac0f0aeb70158c8a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]: /images/20210813/0580946a19714bd9affba9bb24d19a05.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]: /images/20210813/71fe4e79234e4902bd6c4599f85d61b7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]: /images/20210813/aaf2ea3254484f4bbf78ee5507c5160c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]: /images/20210813/4e6bbd60325d4354afd222e9a71ad4ed.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]: /images/20210813/114439f4f7c142629152bde11cbba56e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 7]: /images/20210813/81ce5e61b747411ab6a2294a2ec78827.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 8]: /images/20210813/3b279d31c95542659e934b57c1a6f0c9.png