SpringBoot SpringBoot 开发实用篇 5 整合第三方技术 5.24 SpringBoot 整合 RabbitMQ(topic 模式)

SpringBoot

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

SpringBoot 开发实用篇

5 整合第三方技术

5.24 SpringBoot 整合 RabbitMQ(topic 模式)

5.24.1 整合RabbitMQ 【topic 模式】

之前我们已经使用直连交换机模式【direct 】

完成了消息队列的使用，现在来用一下另一种模式，主题交换机

直接把直连的代码复制一份

等修改完成后，笔者会给出所有的代码

消息服务实现类

package com.dingjiaxiong.service.impl.rabbitmq.topic;

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

/**
 * ClassName: MessageServiceRabbitmqDirectImpl
 * date: 2022/10/22 21:25
 *
 * @author DingJiaxiong
 */

@Service
public class MessageServiceRabbitmqTopicImpl implements MessageService {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Override
    public void sendMessage(String id) {

        System.out.println("待发送短信的订单已纳入处理队列(rabbitmq topic)，id " + id);
        amqpTemplate.convertAndSend("topicExchange","topic.order.id",id);

    }

    @Override
    public String doMessage() {
        return null;
    }
}
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

配置类

package com.dingjiaxiong.service.impl.rabbitmq.topic.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * ClassName: RabbitConfigtopic
 * date: 2022/10/22 21:29
 *
 * @author DingJiaxiong
 */

@Configuration
public class RabbitConfigTopic {

    //存储消息的消息队列的对象
    @Bean
    public Queue topicQueue(){

        return new Queue("topic_queue"); //是否持久化、是否连接专用、是否自动删除【后面其实有三个boolean参数，默认值是tff】
    }

    //还可以再来个队列，让它绑在同一个交换机上
    @Bean
    public Queue topicQueue2(){
        return new Queue("topic_queue2"); //是否持久化、是否连接专用、是否自动删除【后面其实有三个boolean参数，默认值是tff】
    }

    //消息队列不能直接使用，需要使用一个交换机去绑定它
    @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange("topicExchange");
    }

    //做消息队列与交换机的绑定关系
    @Bean
    public Binding bindingTopic(){
        return BindingBuilder.bind(topicQueue()).to(topicExchange()).with("topic.order.id");
    }

    //做消息队列与交换机的绑定关系
    @Bean
    public Binding bindingTopic2(){
        return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("topic2");
    }


}
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
39
40
41
42
43
44
45
46
47
48
49
50
51
52

两个监听器，删掉2，留一个

package com.dingjiaxiong.service.impl.rabbitmq.topic.listener;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

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

@Component
public class MessageListener {

    //收一个消息
    @RabbitListener(queues = "topic_queue")
    public void receive(String id){

        System.out.println("已完成短信发送业务(rabbitmq topic 绑的1号消息队列)，id：" + id);
    }

    //收一个消息
    @RabbitListener(queues = "topic_queue2")
    public void receive2(String id){

        System.out.println("已完成短信发送业务(rabbitmq topic 绑的2号消息队列)，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
24
25
26
27
28
29
30

然后把direct 那套处理一下，全部加载就乱套 了

四个文件全部把注解去掉

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

生产三个

从结果中可以看到，都是1 消费 的

看看服务器

初始化了两个队列

现在有匹配规则要注意

在配置类中可以修改成模糊匹配

这样还是可以匹配上【如果改成topic.*.ids 就不能和后面的匹配上了】

这个和direct 就不同了，direct必须完全匹配，topic模糊匹配的话，就有了一个分发的功能

如果两个队列都能被匹配

先再服务器中把已有的队列删掉

OK，启动服务器进行测试

服务跑起来后，

看看服务器

两个队列也生成了

生产一个消息

效果很明显，两个队列都消费了消息，这说明我们的消息发到了两个队列里

【效果显著，topic 比 direct 直连强大】

回顾一下

• 定义消息队列(topic)

• 绑定键匹配规则
  • * (星号)： 用来表示一个单词 ，且该单词是必须出现的
  • # (井号)： 用来表示任意数量

• 生产与消费消息(topic)

• 使用消息监听器对消息队列监听(topic)

5.24.2 小结

1. SpringBoot整合RabbitMQ主题交换机模式