SpringBoot-RabbitMQ消息的消费与签收机制-天翼云

SpringBoot-RabbitMQ消息的消费与签收机制

2023-06-15 06:23:44 阅读次数：280

消息的签收机制说明

消息消费成功后，我们在客户端签收后，消息就从MQ服务器里面删除了若消息没有消费成功，我们让他回到MQ里面，让别人再次重试消费。

自动签收

消息只要被客户端接收到，无论你客户端发生了什么，我们服务器都不管你了，直接把消息删除了，这是它是默认的行为。

手动签收

创建项目 springboot-rabbitmq，创建方式和之前的方式一样依赖也是。

修改application.yml配置文件：

server:
  port: 8080
spring:
  application:
    name: Springboot-RabbitMQ
  rabbitmq:
    username: user
    password: 123456
    host: 139.196.183.130
    port: 5672
    virtual-host: v-it6666
    # NONE 值是禁用发布确认模式，是默认值
    # CORRELATED 值是发布消息成功到交换机后会触发回调方法
    publisher-confirm-type: correlated
    # 这个是老版本的用法
    # publisher-confirms: true
    # 消息由交换机到达队列失败时触发
    publisher-returns: true
    listener:
      simple:
        # 自动签收，这个是默认行为
        # acknowledge-mode: auto
        # 手动签收
        acknowledge-mode: manual
      direct:
        # 设置直连交换机的签收类型
        acknowledge-mode: manual

消息投递的 ID 说明

SpringBoot-RabbitMQ消息的消费与签收机制

获取投递的 ID

SpringBoot-RabbitMQ消息的消费与签收机制

/**
 * @author BNTang
 */
@Component
public class MessageReceive {

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue,
                    key = {"error"},
                    exchange = @Exchange(value = "directs", type = ExchangeTypes.DIRECT)
            )
    })
    public void receiveMessage(String content, Message message, Channel channel) throws IOException {
        // 消息投递ID
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        // messageId 就是消息的唯一的标识，自己定义
        String messageId = message.getMessageProperties().getMessageId();

        System.out.println("消费者收到消息 → 消息对象：" + message);
        System.out.println("消费者收到消息 → 内容为：" + content);
        System.out.println("消费者收到消息 → 信道：" + channel);
        System.out.println("消息投递ID → ：" + deliveryTag);
        System.out.println("消息自定义ID → ：" + messageId);

        channel.basicAck(deliveryTag, false);
    }
}

basicAck方法参数的解释如下：

deliveryTag：消息投递ID，要签收的投递ID。
multiple：是否批量签收。

投递 ID 存在的问题及消息永久 ID 设置的问题

什么能代表消息的唯一的标识，显然投送的 ID 不行，因为一个消息可能会有多个投送的 ID，我们就需要给消息一个唯一的值，这个伴随消息终身，不会变化！我们需要发送消息时，给消息设置一个 ID，然后保证该 ID 唯一就可以了，如下所示！

@Test
void sendMsg() throws IOException {
    for (int i = 0; i < 5; i++) {
        this.rabbitTemplate.convertAndSend("directs", "error", "我是一个测试消息" + i,
                message -> {
                    String messageId = UUID.randomUUID().toString().replace("-", "");
                    // 自己给消息设置自定义的ID
                    message.getMessageProperties().setMessageId(messageId);
                    return message;
                });
        System.out.println("消息发送成功");
        System.in.read();
    }
}

关于批量签收消息

若我们此时签收了编号为4的消息，但是前面的0，1，2，3 都没有签收，则MQ若是批量的签收，它会把0，1，2，3 都签收，因为MQ认为，比他晚投递的已经签收，前面的肯定已经消费成功了。

生产者

static int a = 1;

@Test
public void sendMessage() throws IOException {
    for (int i = 0; i <= 3; i++) {
        this.rabbitTemplate.convertAndSend("directs", "error", "ABC - " + i, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                // 自己给消息设置自定义的ID
                message.getMessageProperties().setMessageId((a++) + "");
                return message;
            }
        });
    }
    System.out.println("消息发送成功");
    System.in.read();
}

消费者

/**
 * @author BNTang
 */
@Component
public class MessageReceive {

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue("queue"),
                    key = {"error"},
                    exchange = @Exchange(value = "directs")
            )
    })
    public void receiveMessage(String content, Message message, Channel channel) throws IOException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String messageId = message.getMessageProperties().getMessageId();

        System.out.println("消息投递ID → ：" + deliveryTag);
        System.out.println("消息自定义ID → ：" + messageId);

        if (content.equals("ABC - 3")) {
            channel.basicAck(deliveryTag, true);
            System.out.println("消息签收成功 → 内容为：" + content);
        }
    }
}

SpringBoot-RabbitMQ消息的消费与签收机制

可以发现只签收了ABC - 3 但是队列里面没有消息了，说明前面的12都被批量签收了。

不签收

当我们认为消息不合格时，或不是我们要的消息时，我们可以选择不签收它。

生产者

@Test
public void sendMessage() throws IOException {
    this.rabbitTemplate.convertAndSend("directs", "error", "1234567", new MessagePostProcessor() {
        @Override
        public Message postProcessMessage(Message message) throws AmqpException {
            String messageId = UUID.randomUUID().toString().replace("-", "");
            // 自己给消息设置自定义的ID
            message.getMessageProperties().setMessageId(messageId);
            return message;
        }
    });
    System.out.println("消息发送成功");
    System.in.read();
}

消费者

/**
 * @author BNTang
 */
@Component
public class MessageReceive {

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue("queue"),
                    key = {"error"},
                    exchange = @Exchange(value = "directs")
            )
    })
    public void receiveMessage(String content, Message message, Channel channel) throws IOException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String messageId = message.getMessageProperties().getMessageId();

        System.out.println("消息投递ID → ：" + deliveryTag);
        System.out.println("消息自定义ID → ：" + messageId);

        if (content.equals("1234567")) {
            channel.basicAck(deliveryTag, true);
            System.out.println("消息签收成功");
        } else {
            // 如果不是 1234567 就决绝签收
            channel.basicNack(deliveryTag, false, true);
            System.out.println("消息被决绝签收");
        }
    }
}

如上的代码测试方式你先发送一个消息，消息内容为 1234567 这是正常的情况，然后在发送一个 123456 就会发现效果，消息消费死循环了。

我们选择不签收，其实是为了保护消息，当消费消息发生异常时，我们可以把消息放在队列里面，让它重新投递，重新让别人消费！而不是丢了它！

解决不签收消息的死循环

不签收，并且让它回到队列里面，想法很好，但是很容易造成死循环，因为没有任何人能消费她！我们设计一个机制，当一个消息被消费3次还没有消费成功，我们就直接把它记录下来，人工处理！消息消费3次（消息的标识，消息的计数）我们引入Redis，使用Redis计数，若超过3次，直接拒绝消息，并且不回到队列里面。

引入 Redis 依赖，并使用 Docker 运行 Redis，Redis 依赖如下：

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

Docker 运行 Redis 命令脚本如下所示，当然也可以使用本地的Redis图方便，我这里是有机子我就用我的机子了：

docker run -d --name myredis -p 6390:6379 redis --requirepass "1234"

修改消费者的配置文件

SpringBoot-RabbitMQ消息的消费与签收机制

server:
  port: 8002
spring:
  application:
    name: consumer
  rabbitmq:
    host: 139.196.183.130
    port: 5672
    username: user
    password: 123456
    virtual-host: v-it6666
  # Redis的配置
  redis:
    host: 139.196.183.130
    port: 6390
    password: 1234

改造消费者，改造之后的代码如下：

/**
 * @author BNTang
 */
@Component
public class MessageReceive {

    @Autowired
    private StringRedisTemplate redisTemplate;

    /**
     * 消息的前缀
     */
    private String MESSAGE = "message:";

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue("queue"),
                    key = {"error"},
                    exchange = @Exchange(value = "directs")
            )
    })
    public void receiveMessage(String content, Message message, Channel channel) throws IOException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String messageId = message.getMessageProperties().getMessageId();

        System.out.println("消息投递ID → ：" + deliveryTag);
        System.out.println("消息自定义ID → ：" + messageId);

        if (content.equals("1234567")) {
            channel.basicAck(deliveryTag, true);
            System.out.println("消息签收成功");
        } else {
            String count = this.redisTemplate.opsForValue().get(MESSAGE + messageId);

            if (count != null && Long.valueOf(count) >= 3) {
                channel.basicNack(deliveryTag, false, false);
                System.out.println("该消息消费【3】次都失败，我们记录它，人工处理" + content);
            } else {
                // 如果不是 1234567 就决绝签收
                // 处理业务逻辑【可能逻辑处理的出现了问题啥的】
                channel.basicNack(deliveryTag, false, true);
                System.out.println("消息被决绝签收");

                // 因为拒绝了，我们把消息ID放到Redis里面
                this.redisTemplate.opsForValue().increment(MESSAGE + messageId);
            }
        }
    }
}

如上basicNack方法参数的解释如下所示：

deliveryTag：消息的投递ID，要签收的投递ID是多少
multiple：是否批量签收
requeue：true，代表决绝签收，并把消息重新放回队列里面，false，直接拒绝签收

测试注意，因为统计计数时，消息的次数，是通过消息的 ID 来计数的，我们在发送消息时，要设置消息的头：

SpringBoot-RabbitMQ消息的消费与签收机制

活动

应用商城

合作伙伴

开发者

支持与服务

了解天翼云

SpringBoot-RabbitMQ消息的消费与签收机制

SpringBoot-RabbitMQ消息的消费与签收机制

消息的签收机制说明

自动签收

手动签收

消息投递的 ID 说明

获取投递的 ID

投递 ID 存在的问题及消息永久 ID 设置的问题

关于批量签收消息

生产者

消费者

不签收

生产者

消费者

解决不签收消息的死循环

相关文章

redis主从复制集群环境搭建

执行redis-cli命令创建redis集群时报错“Could not connect to Redis at IP:端口: No route to host”

SpringBoot系列之切换log4j日志框架

Redis 实现管道（Pipeline）

SpringCloud-搭建RabbitMQ消息队列

一篇文章教你弄懂redis多模式部署

基于SpringBoot+Vue的宠物领养饲养交流管理平台设计与实现

基于SpringBoot的银行账目账户管理系统设计与实现(源码+lw+部署文档+讲解等)

Redis存储原理与数据模型

在SpringBoot中基于CanvasLabel的地震基础信息展示实践

作者介绍

最新文章

Redis 实现管道（Pipeline）

SpringCloud-搭建RabbitMQ消息队列

基于SpringBoot+Vue的宠物领养饲养交流管理平台设计与实现

基于SpringBoot+Vue的餐饮管理系统设计与实现

SpringCloud-实现基于RabbitMQ的消息队列

基于SpringBoot的高校实习管理系统设计与实现(亮点：多角色、功能全、进可攻退可守)

热门文章

SpringBoot 单元测试(三十六)

(八)高并发redis学习笔记：redis企业级数据备份方案是怎么样的？

（五）高并发redis学习笔记：RDB和AOF两种持久性机制的介绍以及优缺点对比

(七)高并发redis学习笔记：redis的AOF持久化机制深入理解以及数据恢复实验

AOP 中 获取常见的各种参数方法合集

Ajax提交数据SpringBoot后台报错“HttpMessageNotReadableException: JSON parse error: Cannot construct instance“

热门标签

相关产品

弹性云主机

天翼云电脑（公众版）

对象存储

云硬盘

随机文章

C#使用RabbitMQ-5_主题模式（主题交换机）

（九）高并发redis学习笔记：redis主从架构以及读写分离如何承载读高并发

（五）高并发redis学习笔记：RDB和AOF两种持久性机制的介绍以及优缺点对比

高并发核心技术Redis系列(九)--------本地操作

SpringBoot-自动配置原理

redis 锁 PHP防止重复请求

AOP 中获取常见的各种参数方法合集