SpringBoot SpringBoot 开发实用篇 5 整合第三方技术 5.21 SpringBoot 整合 ActiveMQ

SpringBoot

【黑马程序员SpringBoot2全套视频教程，springboot零基础到项目实战（spring boot2完整版）】

SpringBoot 开发实用篇

5 整合第三方技术

5.21 SpringBoot 整合 ActiveMQ

5.21.1 SpringBoot 整合 ActiveMQ

【导入坐标】

<dependency>
    <groupId>org.springframework.bootgroupId>
    <artifactId>spring-boot-starter-activemqartifactId>
dependency>
1
2
3
4

【配置】

server:
  port: 80

spring:
  activemq:
    broker-url: tcp://localhost:61616

  jms:
    template:
      default-destination: dingjiaxiong
1
2
3
4
5
6
7
8
9
10

OK，直接开干，创建真正的实现类

注掉之前那个bean

package com.dingjiaxiong.service.impl.activemq;

import com.dingjiaxiong.service.MessageService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Service;

import java.util.ArrayList;

/**
 * ClassName: MessageServiceActivemqImPL
 * date: 2022/10/22 18:53
 *
 * @author DingJiaxiong
 */

@Service
public class MessageServiceActivemqImpl implements MessageService {

    //模拟一个消息队列
    @Autowired
    private JmsMessagingTemplate messagingTemplate;

    @Override
    public void sendMessage(String id) {
        System.out.println("待发送短信的订单已纳入处理队列，id " + id);
        //把订单id 加入消息队列
        messagingTemplate.convertAndSend(id);
    }

    //从消息队列中取出一条消息进行处理
    @Override
    public String doMessage() {
        String id = messagingTemplate.receiveAndConvert(String.class);
        System.out.println("已完成短信发送业务，id " + id);
        return id;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

直接重启服务器，开始测试

OK， 这样aa 就放进去了，再来个bb

这就是生产消息

下面来消费消息

aa 处理完毕

再发一次

bb 也处理完成

如果再发一次

就会这样一直僵住，但是这个时候我再加一个

厉害了，我这边儿一生产，那边就消费了

【这就实现完毕了一个简单的ActiveMQ 的消息队列】

这个时候还可以查看服务器

点击Home 旁边的Queues

后面两个3，代表我们一共生产了3个消息，消费了3个消息

我们现在再生产一个，不消费

查看控制台

现在就还剩一个没有消费

看看连接

OK，做下来感觉还是挺简单的，有几个地方要提一下

以后每一个消息都发到这一个位置显然不太合理

这样可以自己决定要发到哪儿

举个栗子

重启服务器，生产三个

看看控制台

消费一个

差不多就是这样了，但是实际上，企业开发的时候

不会是说，主动点击什么东西才去消费一下，而是那边儿一生产，这边消费者可以立即消费

MQ 给我们提供了一套这样的机制【监听器】

只要一有消息，我就立马消费

创建一个全新的类

package com.dingjiaxiong.service.impl.activemq.listener;

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

/**
 * ClassName: MessageListener
 * date: 2022/10/22 19:40
 *
 * @author DingJiaxiong
 */

@Component
public class MessageListener {


    //这个方法就可以用来处理消息【而且是监听式的】【一有消息，就能马上调用】
    @JmsListener(destination = "order.queue.id")
    public void receive(String id){
        System.out.println("已完成短信发送业务，id：" + id);
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

直接重启服务测试，【这次的效果应该是发完消息就会马上进行处理】

OK，就是这样，我一生产，它就马上消费

这上面两个是上一次没消费完的

看看控制台

这样就不用“手工消费”了

这样就能实现只要有消息进入队列，就能马上进行处理

【再介绍一个操作】

假如有这样一个场景

这个id 现在处理完后，后面还有一个业务也要使用这个id ，可以实现，它还可以向下转发

package com.dingjiaxiong.service.impl.activemq.listener;

import org.springframework.jms.annotation.JmsListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;

/**
 * ClassName: MessageListener
 * date: 2022/10/22 19:40
 *
 * @author DingJiaxiong
 */

@Component
public class MessageListener {


    //这个方法就可以用来处理消息【而且是监听式的】【一有消息，就能马上调用】
    @JmsListener(destination = "order.queue.id")
    @SendTo("order.other.queue.id")  //这个意思就是说我现在消费完之后，，就把当前方法的返回值放到这个队列中【串起来了】
    public String receive(String id){
        System.out.println("已完成短信发送业务，id：" + id);
        return "new " + id;
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

重启服务器测试

看看控制台

这样我们还可以套娃，继续对新出来那个进行监听，实现一个流水线式的操作

【最后一个问题】

现在我们使用的模型是点对点的

如何更换发布订阅？

配置一下就行了

这样就启用了发布订阅模式

重启服务器测试

生产了四个，消费了四个

看看控制台

滑到下面

欧克欧克，回顾一下

• 导入SpringBoot整合ActiveMQ坐标

• 配置ActiveMQ（采用默认配置）

• 生产与消费消息（使用默认消息存储队列）

• 生产与消费消息（指定消息存储队列）

• 使用消息监听器对消息队列监听

• 流程性业务消息消费完转入下一个消息队列

5.21.2 小结

1. SpringBoot整合ActiveMQ