Spring Boot集成RabbitMQ详解

Joe.Ye • 2023-03-09 • RabbitMQ, Spring Boot

RabbitMQ 即一个消息队列，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。

消息中间件最主要的作用是解耦，中间件最标准的用法是生产者生产消息传送到队列，消费者从队列中拿取消息并处理，生产者不用关心是谁来消费，消费者不用关心谁在生产消息，从而达到解耦的目的。在分布式的系统中，消息队列也会被用在很多其它的方面，比如：分布式事务的支持，RPC 的调用等等。

RabbitMQ 介绍

RabbitMQ 是实现 AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。 RabbitMQ 主要是为了实现系统之间的双向解耦而实现的。当生产者大量产生数据时，消费者无法快速消费，那么需要一个中间层。保存这个数据。

AMQP，即 Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP 的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

RabbitMQ 是一个开源的 AMQP 实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP 等，支持 AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

队列服务

通常我们谈到队列服务, 会有三个概念：发消息者、队列、收消息者，RabbitMQ 在这个基本概念之上, 多做了一层抽象, 在发消息者和队列之间, 加入了交换器 (Exchange). 这样发消息者和队列就没有直接联系, 转而变成发消息者把消息给交换器, 交换器根据调度策略再把消息再给队列。

左侧 P 代表生产者，也就是往 RabbitMQ 发消息的程序。
中间即是 RabbitMQ，其中包括了交换机和队列。
右侧 C 代表消费者，也就是往 RabbitMQ 拿消息的程序。

那么，其中比较重要的概念有 4 个，分别为：虚拟主机，交换机，队列，和绑定。

虚拟主机：一个虚拟主机持有一组交换机、队列和绑定。为什么需要多个虚拟主机呢？很简单， RabbitMQ 当中，用户只能在虚拟主机的粒度进行权限控制。因此，如果需要禁止A组访问B组的交换机/队列/绑定，必须为A和B分别创建一个虚拟主机。每一个 RabbitMQ 服务器都有一个默认的虚拟主机“/”。
交换机：Exchange 用于转发消息，但是它不会做存储，如果没有 Queue bind 到 Exchange 的话，它会直接丢弃掉 Producer 发送过来的消息。这里有一个比较重要的概念：路由键。消息到交换机的时候，交互机会转发到对应的队列中，那么究竟转发到哪个队列，就要根据该路由键。
绑定：也就是交换机需要和队列相绑定，这其中如上图所示，是多对多的关系。

交换机(Exchange)

交换机的功能主要是接收消息并且转发到绑定的队列，交换机不存储消息，在启用ack模式后，交换机找不到队列会返回错误。交换机有四种类型：Direct, topic, Headers and Fanout

Direct：direct 类型的行为是”先匹配, 再投送”. 即在绑定时设定一个routing_key, 消息的routing_key匹配时, 才会被交换器投送到绑定的队列中去.
Topic：按规则转发消息（最灵活）
Headers：设置 header attribute 参数类型的交换机
Fanout：转发消息到所有绑定队列

Direct Exchange

Direct Exchange 是 RabbitMQ 默认的交换机模式，也是最简单的模式，根据key全文匹配去寻找队列。

第一个 X - Q1 就有一个 binding key，名字为 orange； X - Q2 就有 2 个 binding key，名字为 black 和 green。当消息中的路由键和这个 binding key 对应上的时候，那么就知道了该消息去到哪一个队列中。

Ps：为什么 X 到 Q2 要有black，green 2个 binding key呢，一个不就行了吗？ - 这个主要是因为可能又有 Q3，而Q3只接受 black 的信息，而Q2不仅接受black 的信息，还接受 green 的信息。

Topic Exchange

Topic Exchange 转发消息主要是根据通配符。在这种交换机下，队列和交换机的绑定会定义一种路由模式，那么，通配符就要在这种路由模式和路由键之间匹配后交换机才能转发消息。

路由键必须是一串字符，用句号.隔开，比如说 agreements.us，或者 agreements.eu.stockholm 等。
路由模式必须包含一个星号*，主要用于匹配路由键指定位置的一个单词，比如一个路由模式是：agreements..b.*，那么就只能匹配路由键是这样的：第一个单词是 agreements，第四个单词是 b。井号#就表示相当于一个或者多个单词，例如一个匹配模式是 agreements.eu.berlin.#，那么，以agreements.eu.berlin 开头的路由键都是可以的。

具体代码发送的时候还是一样，第一个参数表示交换机，第二个参数表示routing key，第三个参数即消息。如下：

rabbitTemplate.convertAndSend("testTopicExchange", "key1.a.c.key2", " this is  RabbitMQ!");

topic 和 direct 类似, 只是匹配上支持了”模式”, 在”点分”的 routing_key 形式中, 可以使用两个通配符:

*表示一个词
#表示零个或多个词

Headers Exchange

headers 也是根据规则匹配, 相较于 direct 和 topic 固定地使用 routing_key , headers 则是一个自定义匹配规则的类型. 在队列与交换器绑定时, 会设定一组键值对规则, 消息中也包括一组键值对( headers 属性), 当这些键值对有一对, 或全部匹配时, 消息被投送到对应队列.

Fanout Exchange

Fanout Exchange 消息广播的模式，不管路由键或者是路由模式，会把消息发给绑定给它的全部队列，如果配置了 routing_key 会被忽略。

Spring Boot 集成 RabbitMQ

Spring Boot 集成 RabbitMQ 非常简单，如果只是简单的使用配置非常少，Spring Boot 提供了spring-boot-starter-amqp 项目对消息各种支持。

简单使用

（1）配置 Pom 包，主要是添加spring-boot-starter-amqp的支持

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

（2）配置文件

配置 RabbitMQ 的安装地址、端口以及账户信息

spring.application.name=Spring-boot-rabbitmq

spring.rabbitmq.host=192.168.0.86
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest

（3）队列配置

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

@Configuration
public class RabbitConfig {

    @Bean
    public Queue Queue() {
        return new Queue("hello");
    }

}

（4）发送者

rabbitTemplate 是 Spring Boot 提供的默认实现

@Component
public class RabbitSender {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send() {
        String context = "Hello: " + new Date();
        System.out.println("Sender: " + context);
        this.rabbitTemplate.convertAndSend("hello", context);
    }

}

（5）接收者

@Component
@RabbitListener(queues = "hello")
public class RabbitReceiver {

    @RabbitHandler
    public void process(String hello) {
        System.out.println("Receiver: " + hello);
    }

}

（6）测试

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitTest {

    @Autowired
    private RabbitSender rabbitSender;

    @Test
    public void hello() throws Exception {
        rabbitSender.send();
    }

}

注意，发送者和接收者的 queue name 必须一致，不然不能接收

多对多使用

一个发送者，N 个接收者或者 N 个发送者和 N 个接收者会出现什么情况呢？

一对多发送

对上面的代码进行了小改造，接收端注册了两个 Receiver：Receiver1 和 Receiver2，发送端加入参数计数，接收端打印接收到的参数，下面是测试代码，发送一百条消息，来观察两个接收端的执行效果

@Configuration
public class RabbitConfig {

    @Bean
    public Queue manyQueue() {
        return new Queue("many");
    }

}

@Component
@RabbitListener(queues = "many")
public class RabbitReceiver1 {

    @RabbitHandler
    public void process(String content) {
        System.out.println("Receiver1: " + content);
    }

}

@Component
@RabbitListener(queues = "many")
public class RabbitReceiver2 {

    @RabbitHandler
    public void process(String content) {
        System.out.println("Receiver2: " + content);
    }

}

@Component
public class RabbitSender {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send(int i) {
        String content = "Spring Boot Message Queue ****** " + i;
        this.rabbitTemplate.convertAndSend("many", content);
    }

}

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitTest {
    @Autowired
    private RabbitSender rabbitSender;

    @Test
    public void oneToMany() throws Exception {
        for (int i = 0; i < 100; i++) {
            rabbitSender.send(i);
        }
    }
}

结果如下：

Receiver 1: Spring boot Message queue ****** 11
Receiver 2: Spring boot Message queue ****** 12
Receiver 2: Spring boot Message queue ****** 14
Receiver 1: Spring boot Message queue ****** 13
Receiver 2: Spring boot Message queue ****** 15
Receiver 1: Spring boot Message queue ****** 16
Receiver 1: Spring boot Message queue ****** 18
Receiver 2: Spring boot Message queue ****** 17
Receiver 2: Spring boot Message queue ****** 19
Receiver 1: Spring boot Message queue ****** 20

根据返回结果得到以下结论

一个发送者，N个接受者，经过测试会均匀的将消息发送到N个接收者中

多对多发送

复制了一份发送者，加入标记，在一百个循环中相互交替发送

@Test
public void manyToMany() throws Exception {
    for (int i = 0; i < 100; i++) {
        rabbitSender.send(i);
        rabbitSender2.send(i);
    }
}

结论：和一对多一样，接收端仍然会均匀接收到消息

高级使用

对象的支持

Spring Boot 以及完美的支持对象的发送和接收，不需要格外的配置。

@Configuration
public class RabbitConfig {

    @Bean
    public Queue userQueue() {
        return new Queue("user");
    }

}

@Component
public class RabbitSender {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    //发送者
    public void sendUser(User user) {
        System.out.println("Sender object: " + user.toString());
        this.rabbitTemplate.convertAndSend("user", user);
    }

}

@Component
@RabbitListener(queues = {"hello", "many", "user"})
public class RabbitReceiver {

    //接收者
    @RabbitHandler
    public void processUser(User user) {
        System.out.println("Receiver: " + user);
    }

}

@Test
public void objectMessage() throws Exception {
    User user = new User();
    user.setId(1);
    user.setUsername("joe");
    user.setName("Joe.Ye");
    user.setAge(28);
    rabbitSender.sendUser(user);
}

结果如下：

Sender object: User [Hash = 305242558, id=1, username=joe, name=Joe.Ye, age=28, birthday=null]
Receiver object: User [Hash = 1945385112, id=1, username=joe, name=Joe.Ye, age=28, birthday=null]

Topic Exchange

topic 是 RabbitMQ 中最灵活的一种方式，可以根据 routing_key 自由的绑定不同的队列

首先对 topic 规则配置，这里使用两个队列来测试

@Configuration
public class TopicRabbitConfig {

    final static String message = "topic.message";
    final static String messages = "topic.messages";

    @Bean
    public Queue queueMessage() {
        return new Queue(TopicRabbitConfig.message);
    }

    @Bean
    public Queue queueMessages() {
        return new Queue(TopicRabbitConfig.messages);
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange("exchange");
    }

    @Bean
    Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
    }

    @Bean
    Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
    }
}

使用 queueMessages 同时匹配两个队列，queueMessage 只匹配 “topic.message” 队列

@Component
@RabbitListener(queues = "topic.message")
public class TopicRabbitReceiver1 {

    @RabbitHandler
    public void process(String content) {
        System.out.println("Receiver: " + content);
    }

}

@Component
@RabbitListener(queues = "topic.messages")
public class TopicRabbitReceiver2 {

    @RabbitHandler
    public void process(String content) {
        System.out.println("Receiver: " + content);
    }

}

@Component
public class RabbitSender {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void sendTopic1() {
        String content = "Hi, I am message 1";
        System.out.println("Sender: " + content);
        this.rabbitTemplate.convertAndSend("exchange", "topic.message", content);
    }

    public void sendTopic2() {
        String content = "Hi, I am messages 2";
        System.out.println("Sender: " + content);
        this.rabbitTemplate.convertAndSend("exchange", "topic.messages", content);
    }

}

@Test
public void topicMessage() throws Exception {
    //rabbitSender.sendTopic1();
    rabbitSender.sendTopic2();
}

发送sendTopic1会匹配到topic.#和topic.message两个Receiver都可以收到消息，发送sendTopic2只有topic.#可以匹配所有只有Receiver2监听到消息

Fanout Exchange

Fanout 就是我们熟悉的广播模式或者订阅模式，给 Fanout 交换机发送消息，绑定了这个交换机的所有队列都收到这个消息。

Fanout 相关配置

@Configuration
public class FanoutRabbitConfig {
    @Bean
    public Queue AMessage() {
        return new Queue("fanout.A");
    }

    @Bean
    public Queue BMessage() {
        return new Queue("fanout.B");
    }

    @Bean
    public Queue CMessage() {
        return new Queue("fanout.C");
    }

    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }

    @Bean
    Binding bindingExchangeA(Queue AMessage, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(AMessage).to(fanoutExchange);
    }

    @Bean
    Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(BMessage).to(fanoutExchange);
    }

    @Bean
    Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(CMessage).to(fanoutExchange);
    }
}

这里使用了 A、B、C 三个队列绑定到 Fanout 交换机上面，发送端的 routing_key 写任何字符都会被忽略：

@Component
@RabbitListener(queues = "fanout.A")
public class FanoutRabbitReceiver1 {

    @RabbitHandler
    public void process(String content) {
        System.out.println("Receiver A: " + content);
    }

}

@Component
@RabbitListener(queues = "fanout.B")
public class FanoutRabbitReceiver2 {

    @RabbitHandler
    public void process(String content) {
        System.out.println("Receiver B: " + content);
    }

}

@Component
@RabbitListener(queues = "fanout.C")
public class FanoutRabbitReceiver3 {

    @RabbitHandler
    public void process(String content) {
        System.out.println("Receiver C: " + content);
    }

}

public void sendFanout() {
    String content = "Hi, fanout msg";
    System.out.println("Sender: " + content);
    this.rabbitTemplate.convertAndSend("fanoutExchange", "", content);
}

@Test
public void fanoutMessage() throws Exception {
    rabbitSender.sendFanout();
}

结果如下：

Sender: Hi, fanout msg
...
Receiver A: Hi, fanout msg
Receiver C: Hi, fanout msg
Receiver B: Hi, fanout msg

结果说明，绑定到 fanout 交换机上面的队列都收到了消息

版权声明：
作者：Joe.Ye
链接：https://www.appblog.cn/index.php/2023/03/09/spring-boot-integrate-rabbitmq/
来源：APP全栈技术分享
文章版权归作者所有，未经允许请勿转载。

THE END

RabbitMQ

二维码

打赏

海报

Spring Boot集成RabbitMQ详解

RabbitMQ 即一个消息队列，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。消息中间件最主要的作用是解耦，中间件最标准的用法……

Spring Boot内置定时任务

<<上一篇

Spring Boot项目启动时初始化资源

下一篇>>

文章目录

关闭

搜索内容

Spring Boot集成RabbitMQ详解

RabbitMQ 介绍

队列服务

交换机(Exchange)

Direct Exchange

Topic Exchange

Headers Exchange

Fanout Exchange

Spring Boot 集成 RabbitMQ

简单使用

多对多使用

一对多发送

多对多发送

高级使用

对象的支持

Topic Exchange

Fanout Exchange

取消回复

共有 0 条评论

热门文章

最新评论