但行好事
莫论前程❤

Java消息中间件学习笔记(4) — ActiveMQ的使用【队列模式】

JMS代码演练

使用JMS接口规范连接ActiveMQ

创建生产者

创建发布者

创建消费者

创建订阅者

首先回顾一下JMS编码结构之间的关系

img

使用idea创建maven项目

引入依赖:
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.winter.jms</groupId>
    <artifactId>jms-test</artifactId>
    <version>1.0-SNAPSHOT</version>


    <dependencies>
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-all</artifactId>
            <version>5.9.0</version>
        </dependency>
    </dependencies>
</project>
查看项目结构:

img

创建生产者:
package mobi.liupeng.jms.queue;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 *
 * Created by liupeng on 2019/05/15.
 */
public class AppProducer {

    private static final String url = "tcp://127.0.0.1:61616";
    private static final String queueName = "queue-test";

    public static void main(String[] args) throws JMSException {

        //1. 创建ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);

        //2. 创建Connection
        Connection connection = connectionFactory.createConnection();

        //3. 启动连接
        connection.start();

        //4. 创建会话
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        //5. 创建一个目标
        Destination destination = session.createQueue(queueName);

        //6. 创建一个生产者
        MessageProducer producer = session.createProducer(destination);

        for (int i = 0; i < 100; i++) {
            //7. 创建消息
            TextMessage textMessage = session.createTextMessage("test"+i);
            //8. 发布消息
            producer.send(textMessage);
            System.out.println("发送消息"+textMessage.getText());

        }

        //9. 关闭连接
         connection.close();


    }
}

运行查看:

img

在消息队列中已经拥有了100条消息了,此时我们成功的在ActiveMQ中发布了消息。

创建消费者:

注意:其中很多代码都是复制过来的,在实际开发中不可能产生这样的代码,这样是为了使得逻辑清晰而为之

package com.winter.jms.queue;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 *
 * Created by Winterchen on 2017/11/1.
 */
public class AppConsumer {

    private static final String url = "tcp://127.0.0.1:61616";
    private static final String queueName = "queue-test";

    public static void main(String[] args) throws JMSException {

        //1. 创建ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);

        //2. 创建Connection
        Connection connection = connectionFactory.createConnection();

        //3. 启动连接
        connection.start();

        //4. 创建会话
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        //5. 创建一个目标
        Destination destination = session.createQueue(queueName);

       //6. 创建一个消费者
        MessageConsumer consumer = session.createConsumer(destination);

        //7. 创建一个监听器
         consumer.setMessageListener(new MessageListener() {
             public void onMessage(Message message) {
                  TextMessage textMessage = (TextMessage) message;
                 try {
                     System.out.println("接收消息  = [" + textMessage.getText()  + "]");
                 } catch (JMSException e) {

                 }
             }
         });

        //9. 关闭连接
        //connection.close();


    }
}

消费者的连接不允许关闭的,因为消息的接收是异步的,会导致消息不能被消费。  启动运行,可以看到:

接收消息  = [test0]
接收消息  = [test1]
接收消息  = [test2]
接收消息  = [test3]
接收消息  = [test4]
接收消息  = [test5]
接收消息  = [test6]
接收消息  = [test7]
接收消息  = [test8]
接收消息  = [test9]
接收消息  = [test10]
接收消息  = [test11]
接收消息  = [test12]
接收消息  = [test13]
.
.
.

如果我们现在往ActiveMQ中再发布100条消息,然后再启动一个消费者,看看消息的消费情况:

消费者1:

接收消息  = [test0]
接收消息  = [test2]
接收消息  = [test4]
接收消息  = [test6]
接收消息  = [test8]
接收消息  = [test10]
接收消息  = [test12]
接收消息  = [test14]
接收消息  = [test16]
接收消息  = [test18]
接收消息  = [test20]
接收消息  = [test22]
接收消息  = [test24]
接收消息  = [test26]
接收消息  = [test28]
.
.
.

消费者2:

接收消息  = [test1]
接收消息  = [test3]
接收消息  = [test5]
接收消息  = [test7]
接收消息  = [test9]
接收消息  = [test11]
接收消息  = [test13]
接收消息  = [test15]
接收消息  = [test17]
接收消息  = [test19]
接收消息  = [test21]
接收消息  = [test23]
接收消息  = [test25]
接收消息  = [test27]
.
.
.

可以看出,两个消费者是平均消费队列中的所有消息。

赞(0) 打赏
未经允许不得转载:刘鹏博客 » Java消息中间件学习笔记(4) — ActiveMQ的使用【队列模式】
分享到: 更多 (0)

评论 抢沙发

评论前必须登录!

 

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏