zoukankan      html  css  js  c++  java
  • activemq 队列模式(二)

    创建一个maven项目 (quick即可)

    <!--添加activemq的依赖-->
        <dependency>
          <groupId>org.apache.activemq</groupId>
          <artifactId>activemq-all</artifactId>
          <version>5.9.0</version>
        </dependency>
    

     创建生产者:

    package com.test.producermq;
    
    import org.apache.activemq.*;
    
    import javax.jms.*;
    
    /**
     * @Title: MessageProducer
     * @ProjectName activemq
     * @date 2019/11/89:49
     */
    public class MessageProducer {
        //定义ActivMQ的连接地址
        private static final String ACTIVEMQ_URL = "tcp://127.0.0.1:61616";
        //定义发送消息的队列名称
        private static final String QUEUE_NAME = "MyMessage";
    
        public static void main(String[] args) throws JMSException {
            // 1.创建连接工厂
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            // 2.创建连接
            Connection connection = activeMQConnectionFactory.createConnection();
            // 3.打开连接
            connection.start();
            // 4.创建会话
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 5.创建队列目标
            Destination destination = session.createQueue(QUEUE_NAME);
            // 6.创建一个生产者
            javax.jms.MessageProducer producer = session.createProducer(destination);
            //创建模拟100个消息
            for (int i = 1 ; i <= 100 ; i++){
                TextMessage message = session.createTextMessage("我发送message:" + i);
                //发送消息
                producer.send(message);
                //在本地打印消息
                System.out.println("我现在发的消息是:" + message.getText());
            }
            //关闭连接
            connection.close();
        }
    }
    

    运行之后可以看见:

     创建消费者:

    package com.test.consumemq;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    /**
     * @Title: MessageConsumer
     * @ProjectName activemq
     * @date 2019/11/89:56
     */
    public class MessageConsumer {
        //定义ActivMQ的连接地址
        private static final String ACTIVEMQ_URL = "tcp://127.0.0.1:61616";
        //定义发送消息的队列名称
        private static final String QUEUE_NAME = "MyMessage";
        public static void main(String[] args) throws JMSException {
            // 1.创建连接工厂
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            // 2.创建连接
            Connection connection = activeMQConnectionFactory.createConnection();
            // 3.打开连接
            connection.start();
            // 4.创建会话
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 5.创建队列目标
            Destination destination = session.createQueue(QUEUE_NAME);
            // 6.创建消费者
            javax.jms.MessageConsumer consumer = session.createConsumer(destination);
            // 7.创建消费的监听
            consumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    TextMessage textMessage = (TextMessage) message;
                    try {
                        System.out.println("获取消息:" + textMessage.getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }
    

     运行之后可以看见:

      消息已被消费

    队列模式的消息,是只会被一个消费者所使用的,而不会被共享

      

  • 相关阅读:
    GIL 全局解释器
    线程
    队列 Queue 与 生产者消费模型
    进程 与 并发(进程打开方式,互斥锁)
    【JAVA设计模式】单例模式
    【排序】桶排序
    【排序】选择排序
    【排序】插入排序
    【排序】冒泡排序
    JVM 优化之逃逸分析
  • 原文地址:https://www.cnblogs.com/412013cl/p/11818835.html
Copyright © 2011-2022 走看看