专栏首页小灰灰【SpringBoot MQ 系列】RabbitListener 消费基本使用姿势介绍

【SpringBoot MQ 系列】RabbitListener 消费基本使用姿势介绍

之前介绍了 rabbitmq 的消息发送姿势,既然有发送,当然就得有消费者,在 SpringBoot 环境下,消费可以说比较简单了,借助@RabbitListener注解,基本上可以满足你 90%以上的业务开发需求

下面我们来看一下@RabbitListener的最最常用使用姿势

I. 配置

首先创建一个 SpringBoot 项目,用于后续的演示

依赖配置文件 pom.xml

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.2.1.RELEASE</version>
    <relativePath/> <!-- lookup parent from repository -->
</parent>

<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    <java.version>1.8</java.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <!-- 注意,下面这个不是必要的哦-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
</dependencies>

<build>
    <pluginManagement>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </pluginManagement>
</build>
<repositories>
    <repository>
        <id>spring-snapshots</id>
        <name>Spring Snapshots</name>
        <url>https://repo.spring.io/libs-snapshot-local</url>
        <snapshots>
            <enabled>true</enabled>
        </snapshots>
    </repository>
    <repository>
        <id>spring-milestones</id>
        <name>Spring Milestones</name>
        <url>https://repo.spring.io/libs-milestone-local</url>
        <snapshots>
            <enabled>false</enabled>
        </snapshots>
    </repository>
    <repository>
        <id>spring-releases</id>
        <name>Spring Releases</name>
        <url>https://repo.spring.io/libs-release-local</url>
        <snapshots>
            <enabled>false</enabled>
        </snapshots>
    </repository>
</repositories>

application.yml配置文件中,添加 rabbitmq 的相关属性

spring:
  rabbitmq:
    virtual-host: /
    username: admin
    password: admin
    port: 5672
    host: 127.0.0.1

II. 消费姿势

本文将目标放在实用性上,将结合具体的场景来演示@RabbitListener的使用姿势,因此当你发现看完本文之后这个注解里面有些属性还是不懂,请不要着急,下一篇会一一道来

0. mock 数据

消费消费,没有数据,怎么消费呢?所以我们第一步,先创建一个消息生产者,可以往 exchange 写数据,供后续的消费者测试使用

本篇的消费主要以 topic 模式来进行说明(其他的几个模式使用差别不大,如果有需求的话,后续补齐)

@RestController
public class PublishRest {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping(path = "publish")
    public boolean publish(String exchange, String routing, String data) {
        rabbitTemplate.convertAndSend(exchange, routing, data);
        return true;
    }
}

提供一个简单 rest 接口,可以指定往哪个 exchange 推送数据,并制定路由键

1. case1: exchange, queue 已存在

对于消费者而言其实是不需要管理 exchange 的创建/销毁的,它是由发送者定义的;一般来讲,消费者更关注的是自己的 queue,包括定义 queue 并与 exchange 绑定,而这一套过程是可以直接通过 rabbitmq 的控制台操作的哦

所以实际开发过程中,exchange 和 queue 以及对应的绑定关系已经存在的可能性是很高的,并不需要再代码中额外处理;

在这种场景下,消费数据,可以说非常非常简单了,如下:

/**
 * 当队列已经存在时,直接指定队列名的方式消费
 *
 * @param data
 */
@RabbitListener(queues = "topic.a")
public void consumerExistsQueue(String data) {
    System.out.println("consumerExistsQueue: " + data);
}

直接指定注解中的queues参数即可,参数值为对列名(queueName)

2. case2: queue 不存在

当 queue 的 autoDelete 属性为 false 时,上面的使用场景还是比较合适了;但是,当这个属性为 true 时,没有消费者队列就会自动删除了,这个时候再用上面的姿势,可能会得到下面的异常

通常这种场景下,是需要我们来主动创建 Queue,并建立与 Exchange 的绑定关系,下面给出@RabbitListener的推荐使用姿势

/**
 * 队列不存在时,需要创建一个队列,并且与exchange绑定
 */
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "topic.n1", durable = "false", autoDelete = "true"),
        exchange = @Exchange(value = "topic.e", type = ExchangeTypes.TOPIC),
        key = "r"))
public void consumerNoQueue(String data) {
    System.out.println("consumerNoQueue: " + data);
}

一个注解,内部声明了队列,并建立绑定关系,就是这么神奇!!!

注意@QueueBinding注解的三个属性:

  • value: @Queue 注解,用于声明队列,value 为 queueName, durable 表示队列是否持久化, autoDelete 表示没有消费者之后队列是否自动删除
  • exchange: @Exchange 注解,用于声明 exchange, type 指定消息投递策略,我们这里用的 topic 方式
  • key: 在 topic 方式下,这个就是我们熟知的 routingKey

以上,就是在队列不存在时的使用姿势,看起来也不复杂

3. case3: ack

在前面 rabbitmq 的核心知识点学习过程中,会知道为了保证数据的一致性,有一个消息确认机制;

我们这里的 ack 主要是针对消费端而言,当我们希望更改默认 ack 方式(noack, auto, manual),可以如下处理

/**
 * 需要手动ack,但是不ack时
 *
 * @param data
 */
@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "topic.n2", durable = "false", autoDelete = "true"),
        exchange = @Exchange(value = "topic.e", type = ExchangeTypes.TOPIC), key = "r"), ackMode = "MANUAL")
public void consumerNoAck(String data) {
    // 要求手动ack,这里不ack,会怎样?
    System.out.println("consumerNoAck: " + data);
}

上面的实现也比较简单,设置ackMode=MANUAL,手动 ack

但是,请注意我们的实现中,没有任何一个地方体现了手动 ack,这就相当于一致都没有 ack,在后面的测试中,可以看出这种不 ack 时,会发现数据一直在unacked这一栏,当 Unacked 数量超过限制的时候,就不会再消费新的数据了

4. case4: manual ack

上面虽然选择 ack 方式,但是还缺一步 ack 的逻辑,接下来我们看一下如何补齐

/**
 * 手动ack
 *
 * @param data
 * @param deliveryTag
 * @param channel
 * @throws IOException
 */
@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "topic.n3", durable = "false", autoDelete = "true"),
        exchange = @Exchange(value = "topic.e", type = ExchangeTypes.TOPIC), key = "r"), ackMode = "MANUAL")
public void consumerDoAck(String data, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel)
        throws IOException {
    System.out.println("consumerDoAck: " + data);

    if (data.contains("success")) {
        // RabbitMQ的ack机制中,第二个参数返回true,表示需要将这条消息投递给其他的消费者重新消费
        channel.basicAck(deliveryTag, false);
    } else {
        // 第三个参数true,表示这个消息会重新进入队列
        channel.basicNack(deliveryTag, false, true);
    }
}

请注意,方法多了两个参数

  • deliveryTag: 相当于消息的唯一标识,用于 mq 辨别是哪个消息被 ack/nak 了
  • channel: mq 和 consumer 之间的管道,通过它来 ack/nak

当我们正确消费时,通过调用 basicAck 方法即可

// RabbitMQ的ack机制中,第二个参数返回true,表示需要将这条消息投递给其他的消费者重新消费
channel.basicAck(deliveryTag, false);

当我们消费失败,需要将消息重新塞入队列,等待重新消费时,可以使用 basicNack

// 第三个参数true,表示这个消息会重新进入队列
channel.basicNack(deliveryTag, false, true);

5. case5: 并发消费

当消息很多,一个消费者吭哧吭哧的消费太慢,但是我的机器性能又杠杠的,这个时候我就希望并行消费,相当于同时有多个消费者来处理数据

要支持并行消费,如下设置即可

@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "topic.n4", durable = "false", autoDelete = "true"),
        exchange = @Exchange(value = "topic.e", type = ExchangeTypes.TOPIC), key = "r"), concurrency = "4")
public void multiConsumer(String data) {
    System.out.println("multiConsumer: " + data);
}

请注意注解中的concurrency = "4"属性,表示固定 4 个消费者;

除了上面这种赋值方式之外,还有一种 m-n 的格式,表示 m 个并行消费者,最多可以有 n 个

(额外说明:这个参数的解释实在SimpleMessageListenerContainer的场景下的,下一篇文章会介绍它与DirectMessageListenerContainer的区别)

6. 测试

通过前面预留的消息发送接口,我们在浏览器中请求: http://localhost:8080/publish?exchange=topic.e&routing=r&data=wahaha

然后看一下输出,五个消费者都接收到了,特别是主动 nak 的那个消费者,一直在接收到消息;

(因为一直打印日志,所以重启一下应用,开始下一个测试)

然后再发送一条成功的消息,验证下手动真确 ack,是否还会出现上面的情况,请求命令: http://localhost:8080/publish?exchange=topic.e&routing=r&data=successMsg

然后再关注一下,没有 ack 的那个队列,一直有一个 unack 的消息

项目源码

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 【SpringBoot MQ系列教程】RabbitMq 初体验

    mq 在异步解耦削峰的优势非常突出,现在很多的项目都会用到,掌握 mq 的知识点,了解如何顺畅的使用 mq,可以说是一个必备的职业技能点了

    一灰灰blog
  • 【SpringBoot MQ 系列】RabbitMq 消息发送基本使用姿势

    前面两篇博文,分别介绍了RabbitMq的核心知识点,以及整合SpringBoot的demo应用;接下来也该进入正题,看一下SpringBoot的环境下,如何玩...

    一灰灰blog
  • SpringBoot整合RabbitMQ消息队列-学习笔记 原

    本篇文章主要用于记录个人学习RabbitMQ的过程,记录下来方便日后查看,如有错误的地方,还望指正。

    拓荒者
  • 【SpringBoot DB 系列】Redis 高级特性之发布订阅

    通常来讲,当我们业务存在消息的业务逻辑时更多的是直接使用成熟的 rabbitmq,rocketmq,但是一些简单的业务场景中,真的有必要额外的引入一个 mq 么...

    一灰灰blog
  • 【SpringBoot DB 系列】Redis 高级特性之发布订阅

    通常来讲,当我们业务存在消息的业务逻辑时更多的是直接使用成熟的 rabbitmq,rocketmq,但是一些简单的业务场景中,真的有必要额外的引入一个 mq 么...

    一灰灰blog
  • 一起来学SpringBoot | 第十二篇:初探RabbitMQ消息队列

    MQ全称(MessageQueue)又名消息队列,是一种异步通讯的中间件。可以将它理解成邮局,发送者将消息传递到邮局,然后由邮局帮我们发送给具体的消息接收者(消...

    battcn
  • 一起来学 SpringBoot 2.x | 第十二篇:初探 RabbitMQ 消息队列

    rabbitmq基础概念常见应用场景导入依赖属性配置具体编码定义队列实体类控制器消息消费者主函数测试总结说点什么

    芋道源码
  • RabbitMQ六种队列模式之简单队列模式

    RabbitMQ官网提供了七种队列模型,分别是:简单队列、工作队列、发布订阅、路由模式、主题模式、RPC模式,因为RPC模式不是MQ,所以文章系列中不会介绍RP...

    黎明大大
  • 轻松入门SpringBoot整合RabbitMQ

    MQ全称为Message Queue, 消息队列(MQ)是一种应用程序对应用程序的通信方法。MQ是消费-生产者模型的一个典型的代表,一端往消息队列中不断写入消息...

    用户5224393
  • robbitMq 入门学习

    MQ全称为Message Queue, 消息队列(MQ)是一种应用程序对应用程序的通信方法。MQ是消费-生产者模型的一个典型的代表,一端往消息队列中不断写入消息...

    砍鸡鸡
  • 详解SpringCloud中RabbitMQ消息队列原理及配置,一篇就够!

    公众号改版后文章乱序推荐,希望你可以点击上方“Java进阶架构师”,点击右上角,将我们设为★“星标”!这样才不会错过每日进阶架构文章呀。

    java进阶架构师
  • SpringBoot整合RabbitMQ

    1.创建生产者SpringBoot工程 2.导入依赖坐标 3.编写yml配置,基本信息配置 4.定义交换机,队列以及绑定关系的配置类 5.注入Rabbi...

    一只胡说八道的猴子
  • 分布式专题|进入BAT必备之Springboot 集成 RabbitMQ的五种工作模式

    RabbitMQ是由erlang语言开发,基于AMQP(Advanced Message Queue 高级消息队列协议)协议实现的消息队列,它是一种应用程序之间...

    AI码师
  • SpringBoot 整合 Rabbitmq

    场景:用户注册,信息写入数据库后,需要给用户发送注册成功的邮件,再发送注册成功的邮件。

    jwangkun
  • RabbitMQ01-入门介绍

      前面给大家介绍过了ActiveMQ,本文开始给大家介绍下另一个消息中间件RabbitMQ。

    用户4919348
  • @RabbitListener注解你不知道的都在这

    在消息中间件你该了解的秘密一文中详细介绍了如何使用RabbitMQ发送消息、消费消息;如何保证生产者发送消息的可靠性;如何保证消费消息的可靠性、如何横向扩展消费...

    java进阶架构师
  • SpringBoot入门建站全系列(十八)整合RabbitMQ(AMQP类消息队列)

    消息中间件利用高效可靠的消息传递机制进行平台无关的数据交流,并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息排队模型,它可以在分布式环境下扩展进程间...

    品茗IT
  • 『互联网架构』软件架构-Spring boot集成三方中间件(88)

    3.src/main/resources/templates增加error.html

    IT架构圈
  • Spring boot集成三方中间件

    3.src/main/resources/templates增加error.html

    IT架构圈

扫码关注云+社区

领取腾讯云代金券