Java编码实现Activemq通讯

今天药忘吃喽~ 2023-10-11 20:12 0阅读 0赞

**目录**

一、IDEA建Maven工程

二、pom.xml

三、JMS编码总体架构

四、粗说目的地Destination——队列和主题

五、点对点消息传递域中，目的地被称为队列（queue）

六、在发布订阅消息传递域中，目的地被称为主题（topic）

七、topic和queue对比

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

## 一、IDEA建Maven工程 ##

## 二、pom.xml ##

<dependencies>
            
            <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>org.slf4j</groupId>
                <artifactId>slf4j-api</artifactId>
                <version>1.7.30</version>
            </dependency>
            <dependency>
                <groupId>ch.qos.logback</groupId>
                <artifactId>logback-classic</artifactId>
                <version>1.2.11</version>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>1.16.16</version>
            </dependency>
            <dependency>
                <groupId>junit</groupId>
                <artifactId>junit</artifactId>
                <version>4.12</version>
                <scope>test</scope>
            </dependency>
        </dependencies>

## 三、JMS（**Java平台上的专业技术规范）**编码总体架构 ##

![923b24af9e0a41828fc45e277fcffe91.png][]

（1）**回顾JDBC**

![7fd613d06733445c82f1b53d6a734f48.png][]

**（2）JMS（Java平台上的专业技术规范）开发基本步骤**

![ee6d0d7199c44effb4db71e466b5a403.png][]

## 四、粗说目的地Destination——队列和主题 ##

Destination是目的地。下面拿jvm和mq，做个对比。目的地，我们可以理解为是数据存储的地方。

![e50d13698fbb40b49f7302e49f68202b.png][]Destination分为两种：队列和主题。

（1） 队列（queue——1对1）

![068d06a2a7c543ca86313bb986adc8fd.png][]

（2）主题（topic——1对N）

![47becc6b4be54c7f8ac52c90ee75fd12.png][]

## 五、点对点消息传递域中，目的地被称为队列（queue） ##

**（1）消息生产者**

public static final String ACTIVEMQ_URL = "tcp://虚拟机ip: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
            //3.1 两个参数 ①事务 ②签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地（queue or topic）
            Queue queue = session.createQueue(QUEUE_NAME);
            //5.创建生产者
            MessageProducer producer = session.createProducer(queue);
            //6.通过MessageProducer生成3条消息到MQ队列中
            for(int i = 1 ; i <= 3; i ++ ) {
                TextMessage textMessage = session.createTextMessage("msg--" + i);
                producer.send(textMessage);
            }
            producer.close();
            session.close();
            connection.close();
    
            System.out.println("消息发布到MQ完成……");
        }

**（2）控制台说明**

回到ActiveMQ前端控制台可以看到如下界面，有三条消息

![aa6bdc0a9a594487a7fba1d18794f597.png][]

① Number Of Pending Messages：等待消费的消息，这个是未出队列的数量，公式=总接收数-总出队列数。

② Number Of Consumers：消费者数量，消费者端的消费者数量。

③ Messages Enqueued：进队消息数，进队列的总消息量，包括出队列的。这个数只增不减。

④ Messages Dequeued：出队消息数，可以理解为是消费者消费掉的数量。

> 总结：
> 
> 当有一个消息进入这个队列时，等待消费的消息是1，进入队列的消息是1。
> 
> 当消息消费后，等待消费的消息是0，进入队列的消息是1，出队列的消息是1。
> 
> 当再来一条消息时，等待消费的消息是1，进入队列的消息就是2。

**（3）消息消费者**

**① 方式一**

public static final String ACTIVEMQ_URL = "tcp://虚拟机ip: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
            //3.1 两个参数 ①事务 ②签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地（queue or topic）
            Queue queue = session.createQueue(QUEUE_NAME);
            //5.创建消费者
            MessageConsumer consumer = session.createConsumer(queue);
            while(true) {
                //2000ms timeout时间
                //ActiveMQ有五种消息类型，这里发送时是TextMessage，接收时强转一下
                /*同步阻赛方式(receive())
                  订阅者或接收者调Messaoeconsumerreceive()方法来接收消息，receive 方法在能够接收到消息之前(或超时之前)将一直阻塞*/
                TextMessage textMessage = (TextMessage) consumer.receive(2000);
                if(null != textMessage) {
                    System.out.println("接收到消息：" + textMessage.getText());
                } else {
                    break;
                }
            }
            consumer.close();
            session.close();
            connection.close();
        }

![54af63d5ca1943258fec7fdba5f555e4.png][]

> 注意：MessageConsumer的receive方法如果选择的是无参类型的，即不设置超时时间（Timeout），那么上图的Number Of Consumers数量为1，因为消息消费完后receive方法会一直等待，程序一直运行，Consumer也就一直在线

**② 方式二**

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
            //3.1 两个参数 ①事务 ②签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地（queue or topic）
            Queue queue = session.createQueue(QUEUE_NAME);
            //5.创建消费者
            MessageConsumer consumer = session.createConsumer(queue);
            
    
            //通过监听方式读取消息
            /*
            异步非阻塞方式(监听器onMessage())
                订阅者或接收者通过MessageConsumer的setMessageListener(MessageListener     
                listener)注册一个消息监听器当消息到达之后，系统自动调用监听器MessageListener的 
                onMessage(Message message)方法。
            */
            consumer.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();
                        }
                    }
                }
            });
            //连接AvtiveMQ需要等待时间
            System.in.read();
            consumer.close();
            session.close();
            connection.close();
        }

![63458bcce61f4543bb769f949c7ae571.png][]

**③ 消费者情况分析**

1.   **一个消费者**
2.  **两个消费者**
    
    1.  **先生成消息，先启动1号，在启动2号** 结果：1号消费完所有消息，2号等待
    2.  **先启动1，2号，再生成消息** 结果：1，2号均分了消息，颇有些负载均衡的意味

**④ 小总结**

**一、连接的基本步骤**

![9a1a15f2b9004c1d96e43ab2feaa9652.png][]

**二、两种消费方式**

① 同步阻塞方式(receive)

订阅者或接收者抵用MessageConsumer的receive()方法来接收消息，receive方法在能接收到消息之前（或超时之前）将一直阻塞。

② 异步非阻塞方式（监听器onMessage()）

订阅者或接收者通过MessageConsumer的setMessageListener(MessageListener listener)注册一个消息监听器，当消息到达之后，系统会自动调用监听器MessageListener的onMessage(Message message)方法。

**三、队列特点**

![c2b7d2e93dc74d1e8082ab18619c7c84.png][]

## 六、在发布订阅消息传递域中，目的地被称为主题（topic） ##

**（1）topic介绍**

在发布订阅消息传递域中，目的地被称为主题（topic）

发布/订阅消息传递域的特点如下：

① 生产者将消息发布到topic中，每个消息可以有多个消费者，属于1：N的关系；

② 生产者和消费者之间有时间上的相关性。订阅某一个主题的消费者只能消费自它订阅之后发布的消息。

③ 生产者生产时，topic不保存消息它是无状态的不落地，假如无人订阅就去生产，那就是一条废消息，所以，一般**先启动消费者再启动生产者**。

默认情况下如上所述，但是JMS规范允许客户创建持久订阅，这在一定程度上放松了时间上的相关性要求。持久订阅允许消费者消费它在未处于激活状态时发送的消息。一句话，**好比我们的微信公众号订阅。**

![2bbefe07324e457fac2aade48493bc90.png][]

**（2）发布主题生产者**

public static final String ACTIVEMQ_URL = "tcp://虚拟机ip:61616";
        public static final String TOPIC_NAME = "topic01";
    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
            //3.1 两个参数 ①事务 ②签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地（queue or topic）
            Topic topic = session.createTopic(TOPIC_NAME);
            //5.创建生产者
            MessageProducer producer = session.createProducer(topic);
            //6.通过MessageProducer生成3条消息到MQ队列中
            for(int i = 1 ; i <= 3; i ++ ) {
                TextMessage textMessage = session.createTextMessage("topicMessage--" + i);
                producer.send(textMessage);
            }
            producer.close();
            session.close();
            connection.close();
    
            System.out.println("topicMessage消息发布到MQ完成……");
        }

**（3）订阅主题消费者**

public static final String ACTIVEMQ_URL = "tcp://虚拟机ip:61616";
        public static final String TOPIC_NAME = "topic01";
    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
            //3.1 两个参数 ①事务 ②签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地（queue or topic）
            Topic topic = session.createTopic(TOPIC_NAME);
            //5.创建消费者
            MessageConsumer consumer = session.createConsumer(topic);
    
            consumer.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();
                        }
                    }
                }
            });
            //连接AvtiveMQ需要等待时间
            System.in.read();
            consumer.close();
            session.close();
            connection.close();
        }

**（4）先启动消费者再启动生产者**

**① 启动消费者**

![6fd41b0548154935be35a9d03d6d3255.png][]![9836a9a7c28048389318f975551e9ea0.png][]

**② 启动生产者**

![1d7ec66e284748afb87c6e0d8af1b01e.png][]

**③ 消费消息**

![da91e5855e3a49d59ece8b3aadb3c81d.png][]

![1ca8bd3b7a744afda3d85495dd1bb086.png][]

**④ 控制台**

![a9a289a09e1a463d9eae8194363fb9e7.png][]

## 七、topic和queue对比 ##

![96db7d2241694c0696f6f8813cb230dd.png][]

[923b24af9e0a41828fc45e277fcffe91.png]: https://img-blog.csdnimg.cn/923b24af9e0a41828fc45e277fcffe91.png
[7fd613d06733445c82f1b53d6a734f48.png]: https://img-blog.csdnimg.cn/7fd613d06733445c82f1b53d6a734f48.png
[ee6d0d7199c44effb4db71e466b5a403.png]: https://img-blog.csdnimg.cn/ee6d0d7199c44effb4db71e466b5a403.png
[e50d13698fbb40b49f7302e49f68202b.png]: https://img-blog.csdnimg.cn/e50d13698fbb40b49f7302e49f68202b.png
[068d06a2a7c543ca86313bb986adc8fd.png]: https://img-blog.csdnimg.cn/068d06a2a7c543ca86313bb986adc8fd.png
[47becc6b4be54c7f8ac52c90ee75fd12.png]: https://img-blog.csdnimg.cn/47becc6b4be54c7f8ac52c90ee75fd12.png
[aa6bdc0a9a594487a7fba1d18794f597.png]: https://img-blog.csdnimg.cn/aa6bdc0a9a594487a7fba1d18794f597.png
[54af63d5ca1943258fec7fdba5f555e4.png]: https://img-blog.csdnimg.cn/54af63d5ca1943258fec7fdba5f555e4.png
[63458bcce61f4543bb769f949c7ae571.png]: https://img-blog.csdnimg.cn/63458bcce61f4543bb769f949c7ae571.png
[9a1a15f2b9004c1d96e43ab2feaa9652.png]: https://img-blog.csdnimg.cn/9a1a15f2b9004c1d96e43ab2feaa9652.png
[c2b7d2e93dc74d1e8082ab18619c7c84.png]: https://img-blog.csdnimg.cn/c2b7d2e93dc74d1e8082ab18619c7c84.png
[2bbefe07324e457fac2aade48493bc90.png]: https://img-blog.csdnimg.cn/2bbefe07324e457fac2aade48493bc90.png
[6fd41b0548154935be35a9d03d6d3255.png]: https://img-blog.csdnimg.cn/6fd41b0548154935be35a9d03d6d3255.png
[9836a9a7c28048389318f975551e9ea0.png]: https://img-blog.csdnimg.cn/9836a9a7c28048389318f975551e9ea0.png
[1d7ec66e284748afb87c6e0d8af1b01e.png]: https://img-blog.csdnimg.cn/1d7ec66e284748afb87c6e0d8af1b01e.png
[da91e5855e3a49d59ece8b3aadb3c81d.png]: https://img-blog.csdnimg.cn/da91e5855e3a49d59ece8b3aadb3c81d.png
[1ca8bd3b7a744afda3d85495dd1bb086.png]: https://img-blog.csdnimg.cn/1ca8bd3b7a744afda3d85495dd1bb086.png
[a9a289a09e1a463d9eae8194363fb9e7.png]: https://img-blog.csdnimg.cn/a9a289a09e1a463d9eae8194363fb9e7.png
[96db7d2241694c0696f6f8813cb230dd.png]: https://img-blog.csdnimg.cn/96db7d2241694c0696f6f8813cb230dd.png