【微服务】RabbitMQ与SpringAMQP消息队列

发布于：2024-11-28 ⋅ 阅读:(17) ⋅ 点赞:(0)
微信交流群：https://732903.hlcode.cn/?id=NK1fWUR

一、初识MQ 

1. 同步通讯

同步通讯就好比双方打电话，可以实时响应，但只能一对一，只能同时和一个人聊天。

异步通讯就好比两个人发信息，你发信息给对方，对方不一定给你回复，但是可以一对多，可以同时和多个人聊天。

之前学习的Feign远程调用就属于同步方式，虽然调用可以实时得到结果，但存在下面的问题：

同步通讯

优点：

  • 时效性强

缺点:

  • 耦合度高：增加业务需要改动代码
  • 性能和吞吐量低：后面的服务处理需要等待前面的服务结束才能执行，花费时间多
  • 级联失败：前一个服务挂了，后面的服务无法正常运行
  • 有额外的资源消耗

2. 异步通讯

异步调用常见实现就是事件驱动模式，它能够很好的解决同步通讯的弊端。

以购买商品为例，用户支付成功后需要调用订单服务、仓储服务更改订单状态及物流信息。

  • 支付服务是一个事件发布者（publisher），支付成功后发布一个成功的事件(event), 事件携带订单id。
  • 订单服务、仓储服务是事件订阅者（consumer）,订阅支付成功的事件，监听到事件后各自完成自身的使命。
  • 为了降低事件订阅者和事件发布者的耦合度，它们之间有个中介 Broker 。发布者发布事件到Broker，不关心谁来订阅事件。订阅者从Broker订阅事件，不关心谁发来的消息。

异步通讯

优点：

  • 耦合度低： 
  • 吞吐量提升:  调用间没有阻塞，不会造成无效的资源占用
  • 故障隔离
  • 流量削峰: 不管发布了多少事件，都由Broker 接收，订阅者按照自己速度正常执行，可以把 Broker 看作一个缓冲池。

缺点

  • 依赖于Broker 的可靠性、安全性、吞吐能力
  • 架构复杂了，业务没有明显的流程线，不好追踪管理

知道了同步通讯和异步通讯的优缺点，该如何选择呢？

如果业务对于并发性要求不高，但对于时效性要求高，比如要查询某个信息，而且立马需要使用查询的信息。推荐使用同步通讯。

异步通讯适合高并发，性能要求高的场景使用。

3. MQ 常见技术

MQ （MessageQueue），中文是消息队列，字面来看就是存放消息的队列。也就是事件驱动架构中的Broker。

 国内用的比较多的就是RabbitMQ、RocketMQ、Kafka。因为kafka 吞吐量高，可靠性低，所以适用于数据量大，但对数据安全不高的场景使用,如日志。而RabbitMQ和RocketMQ稳定性较强，吞吐量也不差，更适合对稳定性要求高的场景。对于中小型企业，更强调稳定性，可以使用RabbitMQ；如果是大厂，需要做更深入的定制，可以使用RocketMQ。

追求可用性：Kafka、 RocketMQ 、RabbitMQ

追求可靠性：RabbitMQ、RocketMQ

追求吞吐能力：RocketMQ、Kafka

追求消息低延迟：RabbitMQ、Kafka

二、RabbitMQ快速入门

RabbitMQ是基于Erlang语言开发的开源消息通信中间件，官网地址：https://www.rabbitmq.com/

1. 安装RabbitMQ

选择在服务器中使用Docker容器化来安装

  • 下载镜像

  在控制台输入如下指令拉取镜像：

docker pull rabbitmq:3-management
  • 安装MQ

  执行下面的命令来创建并运行rabbitmq容器：

docker run \
 -e RABBITMQ_DEFAULT_USER=zhangsan \
 -e RABBITMQ_DEFAULT_PASS=123456 \
 --name mq \
 --hostname mq1 \
 -p 15672:15672 \
 -p 5672:5672 \
 -d \
 rabbitmq:3-management

上述命令解析： 

  • -e 表示设置环境变量； --name 容器名称； --hostname 主机名，集群用到；
  • RABBITMQ_DEFAULT_USER: MQ管理界面登录账号
  • RABBITMQ_DEFAULT_PASS: MQ管理界面登录密码
  • 15672 ：ui管理平台访问端口
  • 5672 ：内部消息通信端口
  • -d : 后台运行

访问控制台

在浏览器中输入http://192.168.30.130:15672 (服务器ip+端口)，访问并登录后可以看到如下界面。

2. MQ整体架构 

 消息发送者发送消息到交换机，交换机把信息路由到队列中，队列会把信息暂存起来，消息消费者会去消息队列中获取数据。

3. 消息队列模型

常用消息队列模型有5种。基本消息队列(BasicQueue)、工作消息队列(WorkQuez）是基于队列来发送消息的。另外三种又根据交换机类型不同分为三种发布订阅，广播（Fanout Exchange）、路由(Direct Exchange)、主题（Topic  Exchange)。

3.1 基本消息队列模型

官方的HelloWorld是基于最基础的消息队列模型来实现的，只包括三个角色:

  • publisher: 消息发布者，将消息发送到队列queue。
  • queue: 消息队列，负责接受并缓存消息。
  • consumer: 订阅队列，处理队列中的消息。

基本消息队列的消息发送流程

1. 建立connection
2. 创建channel
3. 利用channel声明队列
4. 利用channel向队列发送消息

public class PublisherTest {
    @Test
    public void testSendMessage() throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数，分别是：主机名、端口号(mq内部消息通信端口)、vhost、用户名、密码
        factory.setHost("192.168.30.130");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("zhangsan");
        factory.setPassword("123456");
        // 1.2.建立连接
        Connection connection = factory.newConnection();

        // 2.创建通道Channel
        Channel channel = connection.createChannel();

        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4.发送消息
        String message = "hello, rabbitmq!";
        channel.basicPublish("", queueName, null, message.getBytes());
        System.out.println("发送消息成功：【" + message + "】");

        // 5.关闭通道和连接
        channel.close();
        connection.close();

    }
}


基本消息队列的消息接收流程

1. 建立connection
2. 创建channel
3. 利用channel声明队列
4. 定义consumer的消费行为handleDelivery()
5. 利用channel将消费者与队列绑定

public class ConsumerTest {

    public static void main(String[] args) throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数，分别是：主机名、端口号(mq内部消息通信端口)、vhost、用户名、密码
        factory.setHost("192.168.30.130");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("zhangsan");
        factory.setPassword("123456");
        // 1.2.建立连接
        Connection connection = factory.newConnection();

        // 2.创建通道Channel
        Channel channel = connection.createChannel();

        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4.订阅消息
        channel.basicConsume(queueName, true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 5.处理消息
                String message = new String(body);
                System.out.println("接收到消息：【" + message + "】");
            }
        });
        System.out.println("等待接收消息。。。。");
    }
}

为什么在消息发布中创建队列后，消息接受者也要创建队列呢？

这是因为消息发送者和消息接收者两个的执行顺序是不确定的，有可能前者比后者快，也有可能前者比后者慢。这样是为了保险起见。同个队列，只会创建一次，不必担心创建多次队列。

 

三、SpringAMQP快速入门

1. 概述

官方定义的基本消息队列模型实现还是比较复杂的，因此我们会利用SpringAMQP来实现。想要了解SpringAMQP，就必须先了解什么是AMPQP。

AMPQP 全称为 Advanced Message Queuing Protocol，是用于在应用程序之间传递业务消息的开放标准。该协议与语言和平台无关，更符合微服务中独立性的要求。也就是说AMQP是一种协议，适用于各种开发语言。

Spring AMQP是基于AMQP协议定义的一套API规范，提供了模板来发送和接收消息。包含两部分，其中spring-amgp是基础抽象，spring-rabbit是底层的默认实现。

Spring AMQP官网地址：https://spring.io/projects/spring-amqp

2. BasicQueue 基本消息队列

流程如下

1. 在父工程中引入spring-amqp的依赖。
2. 在publisher服务中利用RabbitTemplate发送消息到simple.queue这个队列。
3. 在consumer服务中编写消费逻辑，绑定simple.queue这个队列。

步骤1：因为publisher和consumer服务都需要amgp依赖，这里把依赖放到父工程mg-demo中。

<!--AMQP依赖，包含RabbitMQ-->
<dependency>    
    <groupId>org.springframework.boot</groupId>    
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

步骤2：在publisher中编写测试方法，向simple.queue发送消息。

  • 在publisher服务中编写application.yml，添加mq连接信息
spring:
  rabbitmq:
    host: 192.168.30.130 # 主机名
    port: 5672 # 端口(mq内部消息通信端口)
    virtual-host: / # 虚拟主机
    username: zhangsan # 用户名
    password: 123456 # 密码
  • 在publisher服务中新建一个测试类，编写测试方法
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAMQPTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSimpleQueue() {
        String queueName = "simple.queue";
        String message = "hello,Spring AMPQ";
        rabbitTemplate.convertAndSend(queueName, message);
    }
}

运行该测试方法，到RabbitMQ管理页面，找到Queue(队列)，进入队列找到 Get messages查看发送的信息。

 

发送信息小结 

  • 引入amgp的starter依赖。因为消息发送者和接受者都要该以来，故放到了父工程依赖上
  • yml 文件上配置RabbitMQ地址
  • 利用RabbitTemplate的convertAndSend方法 

步骤3：在consumer中编写消费逻辑，监听simple.queue。

  • 在consumer服务中编写application.yml，添加mq连接信息
spring:
  rabbitmq:
    host: 192.168.30.130 # 主机名
    port: 5672 # 端口（MQ内部通信端口）
    virtual-host: / # 虚拟主机
    username: zhangsan # 用户名
    password: 123456 # 密码
  • 在consumer服务中新建一个类，编写消费逻辑:

当simple queue队列有消息时，就会来到此类中，接受发送到的消息。一旦消息被接受到了，就会从队列中删除，且不能恢复。

​@Component
public class SpringRabbitListener {

    @RabbitListener(queues = "simple.queue")   // 该注解实时监听消息，queues 指定队列名称
    public void listenSimpleQueueMessage(String msg) {
        System.out.println("Spring 消费者接受到的消息：" + "【" + msg + "】");
    }
}

​

启动该模块启动类。在控制台看到了发送的消息。在MQ管理页面的队列中，消息在队列中删除了。

 

3. WorkQuez 工作消息队列 

如果只有一个消费者，恰好这个消费者处理速度比较慢，每秒钟处理10个消息。队列中的消息来的比较多，每秒发送20个消息，那么多余的十个消息就没有处理，就会滞留在队列中，而队列时在内存中的，总有一个时刻队列会爆满，后面的消息就不能存到队列中，就会出问题。Work queue，工作队列，可以提高消息处理速度，避免队列消息堆积。一条消息只能被其中一个消费者获取。

 

案例：在publisher服务中定义测试方法，每秒产生50条消息，发送到s imple.queue在consumer服务中定义两个消息监听者，都监听simple.queue队列消费者1每秒处理50条消息，消费者2每秒处理10条消息。

步骤1：生产者循环发送消息到simple.queue队列

  • 在publisher服务中添加一个测试方法，循环发送50条消息到simple.queue队列
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAMQPTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;


    @Test
    public void testWorkQueue() throws InterruptedException {
        String queueName = "simple.queue";
        String message = "hello,message_";
        for (int i = 0; i < 50; i++) {
            rabbitTemplate.convertAndSend(queueName, message + i);
        }
        // 避免发送太快
        Thread.sleep(20);
    }

}

步骤2：编写两个消费者，都监听simple.queue

@Component
public class SpringRabbitListener {

    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueueMessage(String msg) throws InterruptedException {
        System.out.println("Spring 消费者1接受到的消息：" + "【" + msg + "】" + LocalTime.now());
        Thread.sleep(25);
    }

    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueueMessage2(String msg) throws InterruptedException {
        System.err.println("Spring 消费者2接受到的消息：" + "【" + msg + "】" + LocalTime.now());
        Thread.sleep(100);
    }
}

步骤3：修改消费者的application.yml文件，设置preFetch这个值，可以控制预取消息的上限

spring:
  rabbitmq:
    host: 192.168.30.130 # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: zhangsan # 用户名
    password: 123456 # 密码
    listener:
      simple:
        prefetch: 1   # 每次只能获取一条消息，处理完成才能获取下一条消息

先启动消息消费者，再启动消息发布者。可以发现能力越大，处理越多。

 

4. FanoutExchange 发布订阅

发布订阅模式与之前案例的区别就是允许将同一消息发送给多个消费者。实现方式是加入了exchange（交换机）。

常见exchange发布订阅类型包括：Fanout ( 广播) ;  Direct (路由) ; Topic (话题)

注意：

exchange负责消息路由，而不是存储，路由失败则消息丢失。

Fanout Exchange 会将接收到的消息广播到每一个跟其绑定的queue。

实现思路： 

  1. 在consumer服务中，利用代码声明队列、交换机，并将两者绑定
  2. 在consumer服务中，编写两个消费者方法，分别监听fanout.queue1和fanout.queue2
  3. 在publisher中编写测试方法，向itcast.fanout发送消息

步骤1：在consumer服务声明Exchange、Queue、Binding

SpringAMQP提供了声明交换机、队列、绑定关系的API

 在消息接受者中创建一个配置类

@Configuration
public class FanoutConfig {

    // 声明交换机
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("zhangsan.fanout");
    }

    // 声明队列1
    @Bean
    public Queue fanoutQueue1() {
        return new Queue("fanout.queue1");
    }

    // 绑定队列1到交换机
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange) {
        return BindingBuilder
                .bind(fanoutQueue1)
                .to(fanoutExchange);
    }

    // 声明队列
    @Bean
    public Queue fanoutQueue2() {
        return new Queue("fanout.queue2");
    }

    // 绑定队列2到交换机
    @Bean
    public Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange) {
        return BindingBuilder
                .bind(fanoutQueue2)
                .to(fanoutExchange);
    }

}

启动Comsumer模块，在RabbitMQ管理页面可以在 Exchanges找到bindings，发现已经绑定了我们定义的队列。

步骤2：在consumer服务声明两个消费者

在consumer服务的SpringRabbitListener类中，添加两个方法，分别监听fanout.queue1和fanout.queue2。

@Component
public class SpringRabbitListener {


    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg) throws InterruptedException {
        System.err.println("消费者接受到fanout.queue1的消息：" + "【" + msg + "】");
        Thread.sleep(100);
    }

    @RabbitListener(queues = "fanout.queue2")
    public void listenFanoutQueue2(String msg) throws InterruptedException {
        System.err.println("消费者接受到fanout.queue2的消息：" + "【" + msg + "】");
        Thread.sleep(100);
    }
}

步骤3：在publisher服务的SpringAmqpTest类中添加测试方法。

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

    @Autowired
    private RabbitTemplate rabbitTemplate;


    @Test
    public void testSendFanoutExchange() {
        // 交换机名
        String exchangeName = "zhangsan.fanout";
        // 消息
        String message = "hello,everyone";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "", message);
    }
}

先启动消息接受者，再启动消息发布者，控制台结果如下。 

5. DirectExchange 发布订阅 

Direct Exchange 将接收到的消息根据规则路由到指定的Queue，因此称为路由模式（routes）。

  • 每一个Queue都与Exchange设置一个BindingKey
  • 发布者发送消息时，指定消息的RoutingKey
  • Exchange将消息路由到BindingKey与消息RoutingKey一致的队列

案例：

步骤1：在consumer服务声明Exchange、Queue 

  • 在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2，
  • 并利用@RabbitListener声明Exchange、Queue、RoutingKey
@Component
public class SpringRabbitListener {


   @RabbitListener(bindings = @QueueBinding(
           value = @Queue(name = "direct.queue1"),
           exchange = @Exchange(name = "zhangsan.direct", type = ExchangeTypes.DIRECT),
           key = {"red", "blue"}
   ))
    public void listenDirectQueue1(String msg) {
        System.err.println("消费者接受到 direct.queue1的消息：" + "【" + msg + "】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue2"),
            exchange = @Exchange(name = "zhangsan.direct", type = ExchangeTypes.DIRECT),
            key = {"red", "yellow"}
    ))
    public void listenDirectQueue2(String msg) {
        System.err.println("消费者接受到 direct.queue2的消息：" + "【" + msg + "】");
    }

}

步骤2：在publisher服务发送消息到DirectExchange

在publisher服务的SpringAmqpTest类中添加测试方法

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

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendDirectExchange() {
        // 交换机名
        String exchangeName = "zhangsan.direct";
        // 消息
        String message = "hello,red";
       // 发送消息, 当路由规则是 red，两个队列都能收到消息，路由规则是blue,队列1收到消息，路由规则是yellow,队列2收到消息
        rabbitTemplate.convertAndSend(exchangeName, "red", message);
    }
}

路由规则是red时 

路由规则时yellow时

6. TopicExchange 发布订阅

TopicExchange与DirectExchange类似，区别在于TopicExchange 的 routingKey必须是多个单词的列表，并且以 . 分割。 DirectExchange的 routingKey是一个单词的列表。

Queue与Exchange指定BindingKey时可以使用通配符：

#：代指0个或多个单词

*：代指一个单词

如下图，如果只关心china的所有信息，bindingKey可以设置成china.#；只关心所有国家的天气 ，bindingKey可以设置成 #.weather, 是不是有一种分组的感觉呢？

 

案例：利用SpringAMQP演示TopicExchange的使用

实现思路如下：

  1. 1. 利用@RabbitListener声明Exchange、Queue、RoutingKey
  2. 2. 在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2
  3. 3. 在publisher中编写测试方法，向itcast. topic发送消息

步骤1： 在consumer服务声明Exchange、Queue

在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2，并利用@RabbitListener声明Exchange、Queue、RoutingKey。

@Component
public class SpringRabbitListener {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue1"),
            exchange = @Exchange(name = "zhangsan.topic", type = ExchangeTypes.TOPIC),
            key = "china.#"
    ))
    public void listenTopicQueue1(String msg) {
        System.out.println("消费者接受到 topic.queue1的消息：" + "【" + msg + "】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue2"),
            exchange = @Exchange(name = "zhangsan.topic", type = ExchangeTypes.TOPIC),
            key = "#.news"
    ))
    public void listenTopicQueue2(String msg) {
        System.out.println("消费者接受到 topic.queue2的消息：" + "【" + msg + "】");
    }


}

步骤2：在publisher服务发送消息到TopicExchange

在publisher服务的SpringAmqpTest类中添加测试方法：

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

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendTopicExchange() {
        // 交换机名
        String exchangeName = "zhangsan.topic";
        // 消息
        String message = "今天我学微服务了";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
    }
}

 7. 扩展（消息转换器）

7.1 引言

在SpringAMQP的发送方法中，接收消息的类型是Object，也就是说我们可以发送任意对象类型的消息，SpringAMQP会帮我们序列化为字节后发送。

下面做个小测试：我们发送一个对象，到MQ管理页面查看是否能发送成功 

步骤1：在consumer中利用@Bean声明一个队列：

@Configuration
public class FanoutConfig {

  // 声明一个队列
    @Bean
    public Queue objectQueue() {
        return new Queue("object.queue");
    }

}

步骤2：在publisher中发送消息以测试：

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

    @Autowired
    private RabbitTemplate rabbitTemplate;

 
    @Test
    public void testSendObject() {
        // 准备消息
        Map<String,Object> msg = new HashMap<>();
        msg.put("name", "Jack");
        msg.put("age", 21);
        // 发送消息
        rabbitTemplate.convertAndSend("object.queue", msg);
    }
}

步骤3：先启动消息接受者，再启动消息发送者，到MQ管理页面，找到Queue队列，查看消息Get Message, 发现我们发送的消息已经序列化了。但这种序列号并不太友好，简单的数据序列化了这么多数据。第一个会影响传输效率；第二个会占用额外的存储空间。

7.2 消息转换器

Spring对消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理的。而默认实现是SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化。如果要修改只需要定义一个MessageConverter 类型的Bean即可。推荐用JSON方式序列化，步骤如下：

消息发送

步骤1：我们在publisher服务引入依赖

  <dependency>
       <groupId>com.fasterxml.jackson.core</groupId>
       <artifactId>jackson-databind</artifactId>
</dependency>

步骤2： publisher服务声明MessageConverter, Bean方法一般都是写在配置类中，这里我写在publisher的启动类中，因为启动类也是个配置类。

 @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

重新启动测试方法，到MQ管理页面，发现我们发送的消息不是序列化的形式了，而是json的形式。

 消息接收

步骤1：在consumer服务引入Jackson依赖

  <dependency>
       <groupId>com.fasterxml.jackson.core</groupId>
       <artifactId>jackson-databind</artifactId>
</dependency>

步骤2：在consumer服务定义MessageConverter , Bean方法一般都是写在配置类中，这里我写在publisher的启动类中，因为启动类也是个配置类。

 @Bean
    public MessageConverter jsonMessageConverter(){
        return new Jackson2JsonMessageConverter();
    }

步骤3：定义一个消费者，监听object.queue队列并消费消息 

@Component
public class SpringRabbitListener {

    @RabbitListener(queues = "object.queue")
    public void listenObjectQueue(Map<String, Object> msg) {
        System.out.println("收到消息：【" + msg + "】"); 
    }
}

步骤4：重启consumer

 

8. 小结

8.1 声明绑定的2种方式

方式1：基于配置类方式声明队列、交换机、绑定队列和交换机

@Configuration
public class FanoutConfig {

    // 声明队列1
    @Bean
    public Queue fanoutQueue1() {
        return new Queue("fanout.queue1");
    }

    // 声明交换机
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("zhangsan.fanout");
    }

    // 绑定队列1到交换机
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange) {
        return BindingBuilder
                .bind(fanoutQueue1)
                .to(fanoutExchange);
    }
}

方式2：基于@RabbitListener注解来声明队列、交换机和绑定队列和交换机的方式

  • Queue：用于声明队列，可以用工厂类QueueBuilder构建
  • Exchange：用于声明交换机，可以用工厂类ExchangeBuilder构建
  • Binding：用于声明队列和交换机的绑定关系，可以用工厂类BindingBuilder构建
@Component
public class SpringRabbitListener {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue1"),
            exchange = @Exchange(name = "zhangsan.topic", type = ExchangeTypes.TOPIC),
            key = "china.#"
    ))
    public void listenTopicQueue1(String msg) {
        System.out.println("消费者接受到 topic.queue1的消息：" + "【" + msg + "】");
    }
}

8.2  五种消息队列的区别

基本消息队列：一个队列绑定一个消费者，消息一旦被使用，立刻删除，不能回溯。

工作队列：一个队列绑定多个消费者，一条消息只被一个消费者使用。

广播发布订阅(FanoutExchange)：发布订阅模式与上面的2种队列区别在于，允许将同一消息发送给多个消费者。不能指定某个队列接收消息。

路由发布订阅(DirectExchange)：路由发布订阅与广播发布订阅的区别在于，路由发布订阅可以根据 routingKey 匹配对应的队列，把消息发送给指定的队列。

话题发布订阅(TopicExchange)：话题发布订阅与路由发布订阅的区别在于，话题发布订阅可以支持模糊匹配，*表示一个，#表示0个或多个。只有符合 routingKey 的队列才能接受到消息。类似于发布朋友圈时哪些分组好友可见一样的效果。

网站公告

今日签到

点亮在社区的每一天
去签到

热门文章

最新发布