【ActiveMQ】之 JMS 开发基本步骤

落日映苍穹つ 2023-02-26 13:28 3阅读 0赞

#### JMS(Java Message Service) 开发基本步骤： ####

1、创建一个 connection factory

2、通过 connection factory 来创建 JMS connection

3、启动 JMS connection

4、通过 connection 创建 JMS session

5、创建 JMS destination 指定是创建 queue 还是 topic

6、创建 JMS producer 或者创建 JMS message 并设置 destination

7、创建 JMS consumer 或者注册一个 JMS message listener

8、发送或者接收 JMS message

9、关闭所有的 JMS 资源（connection, session, producer, consumer 等）

--------------------

#### 具体到 Java 代码如下： ####

**ActiveMQ 编程需要以来如下包：**

<dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-all</artifactId>
        <version>5.15.9</version>
    </dependency>
    <dependency>
        <groupId>org.apache.xbean</groupId>
        <artifactId>xbean-spring</artifactId>
        <version>3.16</version>
    </dependency>
    
    
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>1.7.25</version>
    </dependency>
    <dependency>
        <groupId>ch.qos.logback</groupId>
        <artifactId>logback-classic</artifactId>
        <version>1.2.3</version>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.16.18</version>
        <scope>provided</scope>
    </dependency>

**下面是 JMS producer 的编码实例（使用的 destination 是 queue）：**

public class JmsProducer { 
    
        public static final String ACTIVEMQ_URL = "tcp://localhost:61616";
        public static final String QUEUE_NAME = "my-queue";
    
        @Test
        public void producer() throws JMSException { 
            // 1、创建连接工厂，按照指定的URL地址，采用默认用户名和密码
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
    
            // 2、连接连接工厂获得 connect 连接池并启动访问
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
    
            // 3、创建会话 session，第一个参数叫事务，第二个叫签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
            // 4、创建目的地：队列还是主题 topic
            Queue queue = session.createQueue(QUEUE_NAME);
    
            // 5、创建消息的生产者
            MessageProducer messageProducer = session.createProducer(queue);
    
            // 6、通过生产者生产消息到 MQ 的队列里面
            for (int i = 0; i < 6; i++) { 
                // 7、创建指定格式的消息
                TextMessage textMessage = session.createTextMessage("message ====> " + i);
                // 8、通过 messageProducer 发送消息给 mq
                messageProducer.send(textMessage);
            }
    
            // 9、关闭资源
            messageProducer.close();
            session.close();
            connection.close();
    
            System.out.println("======== 消息发布到 MQ 完成 ==========");
        }
    }

**下面是 JMS consumer 的实例代码：**

public class JmsConsumer { 
    
        public static final String ACTIVEMQ_URL = "tcp://localhost:61616";
        public static final String QUEUE_NAME = "my-queue";
        
        @Test
        public void consumer() throws JMSException, IOException { 
    
            System.out.println(">>>>> 我是消费者");
    
            // 1、创建连接工厂，按照指定的URL地址，采用默认用户名和密码
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
    
            // 2、连接连接工厂获得 connect 连接池并启动访问
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
    
            // 3、创建会话 session，第一个参数叫事务，第二个叫签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
            // 4、创建目的地：队列还是主题 topic
            Queue queue = session.createQueue(QUEUE_NAME);
    
            // 5、创建消息的消费者
            MessageConsumer messageConsumer = session.createConsumer(queue);
    
            // 6、从 MQ 队列中获取消息
            // 使用同步阻塞方法：receive() 会在接收到消息之前或超时之前一直阻塞
            while (true){ 
                TextMessage textMessage = (TextMessage) messageConsumer.receive(3000L);
                if (textMessage != null) { 
                    System.out.println("消费者接收到的消息 ====> " + textMessage.getText());
                } else { 
                    break;
                }
            }
    
            // 7、关闭资源
            messageConsumer.close();
            session.close();
            connection.close();
        }

**其中第6步的从 MQ 队列中获取消息方式可以使用：监听方式**

// 6、从 MQ 队列中获取消息
    // 通过监听方式来获取消息
    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();   // 阻塞等待

**主题 topic 方式也很简单，只需把创建 destination 方式改为 topic 方式即可：**

public static final String TOPIC_NAME = "my-topic";
    
    // 创建目的地为主题 topic
    Topic topic = session.createTopic(TOPIC_NAME);

**当然，生产者和消费者需要保持一致：**

// 创建主题 topic 生产者
    MessageProducer messageProducer = session.createProducer(topic);
    
    // 创建主题 topic 消费者
    MessageConsumer messageConsumer = session.createConsumer(topic);

#### 多消费者问题： ####

**1、生产者先生生产消息，先启动1号消费者再启动2号消费者，结果为：1号消费者可以消费，2号消费者消费不了**  
**2、先启动2个消费者，再生产6条消息，结果为：两个消费者一人一半**