RabbitMQ如何保证队列里的消息99.99%被消费?

Stella981
• 阅读 317

1. 本篇概要

其实,还有1种场景需要考虑:当消费者接收到消息后,还没处理完业务逻辑,消费者挂掉了,那消息也算丢失了?,比如用户下单,订单中心发送了1个消息到RabbitMQ里的队列,积分中心收到这个消息,准备给这个下单的用户增加20积分,但积分还没增加成功呢,积分中心自己挂掉了,导致数据出现问题。

那么如何解决这种问题呢?

为了保证消息被消费者成功的消费,RabbitMQ提供了**消息确认机制(message acknowledgement)**,本文主要讲解RabbitMQ中,如何使用消息确认机制来保证消息被消费者成功的消费,避免因为消费者突然宕机而引起的消息丢失。

2. 开启显式Ack模式

我们开启一个消费者的代码是这样的:

// 创建队列消费者
com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope,
                               AMQP.BasicProperties properties, byte[] body) throws IOException {
        String message = new String(body, "UTF-8");
        System.out.println("Received Message '" + message + "'");
    }
};
channel.basicConsume(QUEUE_NAME, true, consumer);

这里的重点是channel.basicConsume(QUEUE_NAME, true, consumer);方法的第2个参数,让我们先看下basicConsume()的源码:

public String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException {
    return this.basicConsume(queue, autoAck, "", callback);
}

这里的autoAck参数指的是是否自动确认,如果设置为ture,RabbitMQ会自动把发送出去的消息置为确认,然后从内存(或者磁盘)中删除,而不管消费者接收到消息是否处理成功;如果设置为false,RabbitMQ会等待消费者显式的回复确认信号后才会从内存(或者磁盘)中删除。

建议将autoAck设置为false,这样消费者就有足够的时间处理消息,不用担心处理消息过程中消费者宕机造成消息丢失。

此时,队列里的消息就分成了2个部分:

  1. 等待投递给消费者的消息(下图中的Ready部分)
  2. 已经投递给消费者,但是还没有收到消费者确认信号的消息(下图中的Unacked部分)

RabbitMQ如何保证队列里的消息99.99%被消费?

如果RabbitMQ一直没有收到消费者的确认信号,并且消费此消息的消费者已经断开连接,则RabbitMQ会安排该消息重新进入队列,等待投递给下一个消费者,当然也有可能还是原来的那个消费者。

RabbitMQ不会为未确认的消息设置过期时间,它判断此消息是否需要重新投递给消费者的唯一依据是消费该消息的消费者连接是否已经断开,这么设计的原因是RabbitMQ允许消费者消费一条消息的时间可以很久很久。

为了便于理解,我们举个具体的例子,生产者的话的我们延用上文中的DurableProducer:

package com.zwwhnly.springbootaction.rabbitmq.durable;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class DurableProducer {
    private final static String EXCHANGE_NAME = "durable-exchange";
    private final static String QUEUE_NAME = "durable-queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接
        ConnectionFactory factory = new ConnectionFactory();
        // 设置 RabbitMQ 的主机名
        factory.setHost("localhost");
        // 创建一个连接
        Connection connection = factory.newConnection();
        // 创建一个通道
        Channel channel = connection.createChannel();
        // 创建一个Exchange
        channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

        // 发送消息
        String message = "durable exchange test";
        AMQP.BasicProperties props = new AMQP.BasicProperties().builder().deliveryMode(2).build();
        channel.basicPublish(EXCHANGE_NAME, "", props, message.getBytes());

        // 关闭频道和连接
        channel.close();
        connection.close();
    }
}

然后新建一个消费者AckConsumer类:

package com.zwwhnly.springbootaction.rabbitmq.ack;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class AckConsumer {
    private final static String QUEUE_NAME = "durable-queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接
        ConnectionFactory factory = new ConnectionFactory();
        // 设置 RabbitMQ 的主机名
        factory.setHost("localhost");
        // 创建一个连接
        Connection connection = factory.newConnection();
        // 创建一个通道
        Channel channel = connection.createChannel();
        // 创建队列消费者
        com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                int result = 1 / 0;
                System.out.println("Received Message '" + message + "'");
            }
        };
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

我们先将autoAck参数设置为ture,即自动确认,并在消费消息时故意写个异常,然后先运行生产者客户端将消息写入队列中,然后运行消费者客户端,发现消息未消费成功但是却消失了:

RabbitMQ如何保证队列里的消息99.99%被消费?

RabbitMQ如何保证队列里的消息99.99%被消费?

然后我们将autoAck设置为false:

channel.basicConsume(QUEUE_NAME, false, consumer);

再次运行生产者客户端将消息写入队列中,然后运行消费者客户端,此时虽然消费者客户端仍然代码异常,但是消息仍然在队列中:

RabbitMQ如何保证队列里的消息99.99%被消费?

然后我们删除掉消费者客户端中的异常代码,重新启动消费者客户端,发现消息消费成功了,但是消息一直未Ack:

RabbitMQ如何保证队列里的消息99.99%被消费?

RabbitMQ如何保证队列里的消息99.99%被消费?

手动停掉消费者客户端,发现消息又到了Ready状态,准备重新投递:

RabbitMQ如何保证队列里的消息99.99%被消费?

之所以消费掉消息,却一直还是Unacked状态,是因为我们没在代码中添加显式的Ack代码:

String message = new String(body, "UTF-8");
//int result = 1 / 0;
System.out.println("Received Message '" + message + "'");

long deliveryTag = envelope.getDeliveryTag();
channel.basicAck(deliveryTag, false);

deliveryTag可以看做消息的编号,它是一个64位的长整形值。

此时运行消费者客户端,发现消息消费成功,并且在队列中被移除: RabbitMQ如何保证队列里的消息99.99%被消费? RabbitMQ如何保证队列里的消息99.99%被消费?

点关注,不迷路,这是一个程序员都想要关注的二维码

RabbitMQ如何保证队列里的消息99.99%被消费?

点赞
收藏
评论区
推荐文章
秃头王路飞 秃头王路飞
2个月前
webpack5手撸vue2脚手架
webpack5手撸vue相信工作个12年的小伙伴们在面试的时候多多少少怕被问到关于webpack方面的知识,本菜鸟最近闲来无事,就尝试了手撸了下vue2的脚手架,第一次发帖实在是没有经验,望海涵。languageJavaScript"name":"vuecliversion2","version":"1.0.0","desc
刚刚好 刚刚好
2个月前
css问题
1、在IOS中图片不显示(给图片加了圆角或者img没有父级)<div<imgsrc""/</divdiv{width:20px;height:20px;borderradius:20px;overflow:h
blmius blmius
1年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
晴空闲云 晴空闲云
2个月前
css中box-sizing解放盒子实际宽高计算
我们知道传统的盒子模型,如果增加内边距padding和边框border,那么会撑大整个盒子,造成盒子的宽度不好计算,在实务中特别不方便。boxsizing可以设置盒模型的方式,可以很好的设置固定宽高的盒模型。盒子宽高计算假如我们设置如下盒子:宽度和高度均为200px,那么这会这个盒子实际的宽高就都是200px。但是当我们设置这个盒子的边框和内间距的时候,那
艾木酱 艾木酱
1个月前
快速入门|使用MemFire Cloud构建React Native应用程序
MemFireCloud是一款提供云数据库,用户可以创建云数据库,并对数据库进行管理,还可以对数据库进行备份操作。它还提供后端即服务,用户可以在1分钟内新建一个应用,使用自动生成的API和SDK,访问云数据库、对象存储、用户认证与授权等功能,可专
Stella981 Stella981
1年前
KVM调整cpu和内存
一.修改kvm虚拟机的配置1、virsheditcentos7找到“memory”和“vcpu”标签,将<namecentos7</name<uuid2220a6d1a36a4fbb8523e078b3dfe795</uuid
Wesley13 Wesley13
1年前
MySQL查询按照指定规则排序
1.按照指定(单个)字段排序selectfromtable_nameorderiddesc;2.按照指定(多个)字段排序selectfromtable_nameorderiddesc,statusdesc;3.按照指定字段和规则排序selec
Wesley13 Wesley13
1年前
JMS消息的概念解释
1、默认生产者消息是持久的:会存数据库\消费者的持久:createDurableSubscriber是指消费者能收到所有它订阅时间点之后的消息,即使消费者注册后关闭,当它重启就能收到注册时间点之后所有的消息;即当此消费用户ID(AAA)在producer发送之前就已经注册,那么此id能收到producer发送的所有消息,如果是在produce
Wesley13 Wesley13
1年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
helloworld_34035044 helloworld_34035044
5个月前
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为
helloworld_28799839 helloworld_28799839
2个月前
常用知识整理
Javascript判断对象是否为空jsObject.keys(myObject).length0经常使用的三元运算我们经常遇到处理表格列状态字段如status的时候可以用到vue