Java编程实现ActiveMQ的队列

忘是亡心i 2023-06-18 07:59 3阅读 0赞

# 编程实现\[队列\] #

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

## 生产者 ##

1、创建一个Maven工程，使用quick骨架即可  
2、引入依赖

<dependency>
          <groupId>org.apache.activemq</groupId>
          <artifactId>activemq-all</artifactId>
          <version>5.9.0</version>
        </dependency>

3、编程实现

package cn.oceanstar;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    /**
     * Hello world!
     *
     */
    public class App 
    {
        
    
        private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;    //默认连接用户名
        private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD; //默认连接密码
        private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;     //默认连接地址
    
        public static final String QUEUQ_NAME = "queue01";
    
        public static void main(String[] args) throws Exception{
        
    
    
            //1、使用默认用户名和密码创建连接工厂
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(App.USERNAME, App.PASSWORD, App.BROKEURL);;
    
            //2、获取connection, 并启动访问
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
    
            //3、创建会话： 两个参数：事务, 签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
            //4、创建目的地：主题/队列
            Queue queue = session.createQueue(QUEUQ_NAME);
    
            //5、消息生产者
            MessageProducer messageProducer = session.createProducer(queue);
    
            //6、消息生产者生产消息并发送到mq的队列里面
            for (int i=1;i <=3; i++){
        
                //7、创建消息
                TextMessage textMessage = session.createTextMessage("msg:" + i);
    
                //8、发送消息
                messageProducer.send(textMessage);
            }
    
            //9、关闭资源
            messageProducer.close();
            session.close();
            connection.close();
    
            System.out.printf("消息发送完成了。。。。");
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][20190925123737523.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70 2]  
消费者数量：表示正在连接这个队列的消费者。

## 消费者 ##

package cn.oceanstar;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    public class JmsConsumer {
        
        private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;    //默认连接用户名
        private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD; //默认连接密码
        private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;     //默认连接地址
    
        public static final String QUEUQ_NAME = "queue01";
    
        public static void main(String[] args) throws Exception {
        
            //1、使用默认用户名和密码创建连接工厂
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(JmsConsumer.USERNAME, JmsConsumer.PASSWORD, JmsConsumer.BROKEURL);;
    
            //2、获取connection, 并启动访问
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
    
            //3、创建会话： 两个参数：事务, 签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
            //4、创建目的地：主题/队列
            Queue queue = session.createQueue(QUEUQ_NAME);
    
            //5、创建消费者
            MessageConsumer messageConsumer = session.createConsumer(queue);
    
            //-------------
            /*
           同步阻塞方式:receive()
            在接受到消息之前(或者超时之前)将一直阻塞
            while (true){
                //TextMessage textMessage =  (TextMessage) messageConsumer.receive(); // 一直等
                TextMessage textMessage =  (TextMessage) messageConsumer.receive(4*1000); //等4s，过失不候
                if (null != textMessage){
                    System.out.println("消费者收到消息：" + textMessage.getText());
                }else {
                    break;
                }
            }
            */
            //-------------------------
            //------------------------------------------
           //通过监听的方式来消费消息
            messageConsumer.setMessageListener(new MessageListener() {
          //注册监听器
                @Override  //消息到达之后，系统自动调用监听器MessageListener的onMessage
                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();
        }
    }

![在这里插入图片描述][20190925130501246.png]  
![在这里插入图片描述][20190925130512179.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70 3]

## 总结 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70 4]  
队列是点对点的。点对点消息传递域的特点如下：

*  每个消息只能有一个消费者，是一对一的。
 *  消息的生产者和消费者之间没有时间上的相关性
 *  消息被消费后队列中不会再存储，所以消费者不会消费到已经被消费掉的消息。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2019092512562750.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20190925115218379.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg==,size_16,color_FFFFFF,t_70
[20190925123737523.png]: https://img-blog.csdnimg.cn/20190925123737523.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/201909251252272.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg==,size_16,color_FFFFFF,t_70
[20190925130501246.png]: https://img-blog.csdnimg.cn/20190925130501246.png
[20190925130512179.png]: https://img-blog.csdnimg.cn/20190925130512179.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20190925133329584.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20190925133555743.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg==,size_16,color_FFFFFF,t_70