当前位置： 首页 > news >正文

RabbitMQ消息可靠性保证机制3--消费端ACK机制

news 来源：原创 2024/5/17 10:38:14

消费端ACK机制

​ 在这之前已经完成了发送端的确认机制。可以保证数据成功的发送到RabbitMQ，以及持久化机制，然尔这依然无法完全保证整个过程的可靠性，因为如果消息被消费过程中业务处理失败了，但是消息却已经被标记为消费了，如果又没有任何重度机制，那结果基本等于丢消息。在消费端如何保证消息不丢呢？

在rabbitMQ的消费端会有ACK机制。即消费端消费消息后需要发送ACK确认报文给Broker端，告知自己是否已经消费完成，否则可能会一直重发消息直到消息过期（AUTO模式）。同时这个也是最终一致性、可恢复性的基础。一般有如下手段：
  1. 采用NONE模式，消费的过程中自行捕捉异常，引发异常后直接记录日志并落到异常处理表，再通过后台定时任务扫描异常恢复表做重度动作。如果业务不自行处理则有丢失数据的风险。
  2. 采用AUTO（自动ACK）模式，不主动捕获异常，当消费过程中出现异常时，会将消息放回Queue中，然后消息会被重新分配到其他消费节点（如果没有则还是选择当前节点）重新被消费，默认会一直重发消息并直到消费完成返回ACK或者一直到过期。
  3. 采用MANUAL（手动ACK）模式，消费者自行控制流程并手动调用channel相关的方法返回ACK。
7.6.1 手动ACK机制-Reject

maven导入

            <dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.9.0</version></dependency>

生产者

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.nio.charset.StandardCharsets;public class Product {public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setUri("amqp://root:123456@node1:5672/%2f");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel(); ) {// 定义交换器队列channel.exchangeDeclare("ack.ex", BuiltinExchangeType.DIRECT, false, false, false, null);// 定义队列channel.queueDeclare("ack.qu", false, false, false, null);// 队列绑定channel.queueBind("ack.qu", "ack.ex", "ack.rk");for (int i = 0; i < 5; i++) {byte[] sendBytes = ("hello-" + i).getBytes(StandardCharsets.UTF_8);channel.basicPublish("ack.ex", "ack.rk", null, sendBytes);}} catch (Exception e) {e.printStackTrace();}}
}

消费者

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException;public class Consumer {public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setUri("amqp://root:123456@node1:5672/%2f");Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare("ack.qu", false, false, false, null);DefaultConsumer consumer =new DefaultConsumer(channel) {@Overridepublic void handleDelivery(// 消费者标签String consumerTag,// 消费者的封装Envelope envelope,// 消息属性AMQP.BasicProperties properties,// 消息体byte[] body)throws IOException {System.out.println("确认的消息内容:" + new String(body));// 找收消息// Nack与Reject的区别在于，nack可以对多条消息进行拒收，而reject只能拒收一条。// requeue为true表示不确认的消息会重新放回队列。channel.basicReject(envelope.getDeliveryTag(), true);}};channel.basicConsume("ack.qu",// 非自动确认false,// 消费者的标签"ack.consumer",// 回调函数consumer);}
}

发送测试

首先执行生产者向队列中发送数据。然后执行消费者，检查拒收的处理。

在消费者的控制台，将持续不断的输出消息信息：

确认的消息内容:hello-0
确认的消息内容:hello-1
确认的消息内容:hello-2
确认的消息内容:hello-3
确认的消息内容:hello-4
确认的消息内容:hello-0
确认的消息内容:hello-1
......
确认的消息内容:hello-0

按照发送的顺序将不断的被打印。

那此时消息是什么状态呢？查看下消息队列中的信息

[root@nullnull-os rabbitmq]# rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers  --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name          │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu        │ 0              │ 5                       │ 5        │ 1         │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1              │ 0                       │ 1        │ 0         │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘
[root@nullnull-os rabbitmq]#

可以看到当前的消息处于unack的状态。由于消息被不断的重新放回队列，而消费者又只有当前这一个，所以，在不断拒收中被放回。

那如果将消息拒绝改为不重新放回队列，会如何呢？来验证下。

首先修改消费者的代码:

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException;public class Consumer {public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setUri("amqp://root:123456@node1:5672/%2f");Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare("ack.qu", false, false, false, null);DefaultConsumer consumer =new DefaultConsumer(channel) {@Overridepublic void handleDelivery(// 消费者标签String consumerTag,// 消费者的封装Envelope envelope,// 消息属性AMQP.BasicProperties properties,// 消息体byte[] body)throws IOException {System.out.println("确认的消息内容:" + new String(body));// 找收消息// Nack与Reject的区别在于，nack可以对多条消息进行拒收，而reject只能拒收一条。// requeue为false表示不确认的消息不会重新放回队列。//channel.basicReject(envelope.getDeliveryTag(), true);channel.basicReject(envelope.getDeliveryTag(), false);}};channel.basicConsume("ack.qu",// 非自动确认false,// 消费者的标签"ack.consumer",// 回调函数consumer);}
}

再次执行消费者。

确认的消息内容:hello-0
确认的消息内容:hello-1
确认的消息内容:hello-2
确认的消息内容:hello-3
确认的消息内容:hello-4

而这一次消息没有再循环打印。只输出一遍，再检查下消息在队列中的状态：

[root@nullnull-os rabbitmq]# rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers  --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name          │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu        │ 0              │ 0                       │ 0        │ 1         │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1              │ 0                       │ 1        │ 0         │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘
[root@nullnull-os rabbitmq]#

通过观察发现，消息已经没有在队列中了，那就是消息已经被丢弃了。

7.6.2 手动ACK机制-ack

消费者修改为ACK确认处理

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;import java.io.IOException;public class Consumer {public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setUri("amqp://root:123456@node1:5672/%2f");Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare("ack.qu", false, false, false, null);DefaultConsumer consumer =new DefaultConsumer(channel) {@Overridepublic void handleDelivery(// 消费者标签String consumerTag,// 消费者的封装Envelope envelope,// 消息属性AMQP.BasicProperties properties,// 消息体byte[] body)throws IOException {System.out.println("确认的消息内容:" + new String(body));// 消息确认，并且非批量确认，multiple为false，表示只确认了单条channel.basicAck(envelope.getDeliveryTag(), false);}};channel.basicConsume("ack.qu",// 非自动确认false,// 消费者的标签"ack.consumer",// 回调函数consumer);}
}

此时可以先运行消息者。等待消息推送。然后运行生产者将消息推送，此时便可以看到消费者的控制台输出：

确认的消息内容:hello-0
确认的消息内容:hello-1
确认的消息内容:hello-2
确认的消息内容:hello-3
确认的消息内容:hello-4

观察队列中的信息

[root@nullnull-os rabbitmq]# rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers  --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name          │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu        │ 0              │ 0                       │ 0        │ 1         │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1              │ 0                       │ 1        │ 0         │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘
[root@nullnull-os rabbitmq]#

在队列中，消息已经被成功的消费了。

7.6.3 手动ACK机制-nack
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException;public class Consumer {public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setUri("amqp://root:123456@node1:5672/%2f");Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare("ack.qu", false, false, false, null);DefaultConsumer consumer =new DefaultConsumer(channel) {@Overridepublic void handleDelivery(// 消费者标签String consumerTag,// 消费者的封装Envelope envelope,// 消息属性AMQP.BasicProperties properties,// 消息体byte[] body)throws IOException {System.out.println("确认的消息内容:" + new String(body));// 消息批量不确认，即批量丢弃，每5个做一次批量消费// 参数1，消息的标签// multiple为false 表示不确认当前是一个消息。true就是多个消息。// requeue为true表示不确认的消息会重新放回队列。// 每5条做一次批量确认,_deliveryTag从1开始if (envelope.getDeliveryTag() % 5 == 0) {System.out.println("批量确认执行");channel.basicNack(envelope.getDeliveryTag(), true, false);}}};channel.basicConsume("ack.qu",// 非自动确认false,// 消费者的标签"ack.consumer",// 回调函数consumer);}
}

执行消费者程序,然后再执行生产者。查看消费端的控制台：

确认的消息内容:hello-0
确认的消息内容:hello-1
确认的消息内容:hello-2
确认的消息内容:hello-3
确认的消息内容:hello-4
批量确认执行

由于此处采用的是不重新放回队列，所以，数据接收到之后被丢弃了。

[root@nullnull-os rabbitmq]# rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers  --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name          │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu        │ 0              │ 0                       │ 0        │ 0         │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1              │ 0                       │ 1        │ 0         │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘

队列中的数据也已经被处理掉了。

7.6.4 手动ACK机制-SpringBoot

首先是Maven导入

        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId><version>2.2.8.RELEASE</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId><version>2.2.8.RELEASE</version></dependency>

配制文件application.yml

spring:application:name: consumer-ackrabbitmq:host: node1port: 5672virtual-host: /username: rootpassword: 123456# 配制消费端ack信息。listener:simple:acknowledge-mode: manual# 重试超过最大次数后是否拒绝default-requeue-rejected: falseretry:# 开启消费者重度(false时关闭消费者重试，false不是不重试，而是一直收到消息直到ack确认或者一直到超时）enable: true# 最大重度次数max-attempts: 5# 重试间隔时间(单位毫秒)initial-interval: 1000

主启动类

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;import java.nio.charset.StandardCharsets;@SpringBootApplication
public class Main {@Autowired private RabbitTemplate rabbitTemplate;public static void main(String[] args) {SpringApplication.run(Main.class, args);}/*** 在启动后就开始向MQ中发送消息** @return*/@Beanpublic ApplicationRunner runner() {return args -> {Thread.sleep(5000);for (int i = 0; i < 10; i++) {MessageProperties props = new MessageProperties();props.setDeliveryTag(i);Message message = new Message(("消息:" + i).getBytes(StandardCharsets.UTF_8), props);rabbitTemplate.convertAndSend("ack.ex", "ack.rk", message);}};}
}

当主类启动后，会延迟5秒，向MQ中发送10条记录。

队列配制

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitConfig {@Beanpublic Queue queue() {return new Queue("ack.qu", false, false, false, null);}@Beanpublic Exchange exchange(){return new DirectExchange("ack.ex",false,false,null);}@Beanpublic Binding binding(){return BindingBuilder.bind(queue()).to(exchange()).with("ack.rk").noargs();}
}

使用推送模式来查确认消息

监听器，MQ队列推送消息至listener

import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;import java.io.IOException;
import java.util.concurrent.ThreadLocalRandom;@Component
public class MessageListener {/*** NONE模式，则只要收到消息后就立即确认（消息出列，标识已消费），有丢数据风险** <p>AUTO模式，看情况确认，如果此时消费者抛出异常则消息会返回队列中** <p>WANUAL模式，需要显示的调用当前channel的basicAck方法** @param channel* @param deliveryTag* @param msg*/// @RabbitListener(queues = "ack.qu", ackMode = "AUTO")// @RabbitListener(queues = "ack.qu", ackMode = "NONE")@RabbitListener(queues = "ack.qu", ackMode = "MANUAL")public void handMessageTopic(Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, @Payload String msg) {System.out.println("消息内容：" + msg);ThreadLocalRandom current = ThreadLocalRandom.current();try {if (current.nextInt(10) % 3 != 0) {// 手动nack，告诉broker消费者处理失败，最后一个参数表示是否需要将消息重新入列// channel.basicNack(deliveryTag, false, true);// 手动拒绝消息，第二个参数表示是否重新入列channel.basicReject(deliveryTag, true);} else {// 手动ACK，deliveryTag表示消息的唯一标志，multiple表示是否批量确认channel.basicAck(deliveryTag, false);System.out.println("已经确认的消息" + msg);}Thread.sleep(ThreadLocalRandom.current().nextInt(500, 3000));} catch (IOException e) {e.printStackTrace();} catch (InterruptedException e) {}}
}

消息有33%的概率被拒绝，这样又会被重新放回队列，等待下次推送。

启动测试

运行main方法

【确认】消息内容：消息:0
【拒绝】消息内容：消息:1
【拒绝】消息内容：消息:2
【拒绝】消息内容：消息:3
【确认】消息内容：消息:4
【确认】消息内容：消息:5
【拒绝】消息内容：消息:6
【拒绝】消息内容：消息:7
【拒绝】消息内容：消息:8
【拒绝】消息内容：消息:9
【确认】消息内容：消息:1
【拒绝】消息内容：消息:2
【拒绝】消息内容：消息:3
【拒绝】消息内容：消息:6
【确认】消息内容：消息:7
【确认】消息内容：消息:8
【拒绝】消息内容：消息:9
【拒绝】消息内容：消息:2
【拒绝】消息内容：消息:3
【拒绝】消息内容：消息:6
【确认】消息内容：消息:9
【确认】消息内容：消息:2
【拒绝】消息内容：消息:3
【拒绝】消息内容：消息:6
【确认】消息内容：消息:3
【拒绝】消息内容：消息:6
【确认】消息内容：消息:6

从观察到的结果也印证了，反复的被推送，接收的一个过程中，使用命令查看队列的一个消费的情况

[root@nullnull-os rabbitmq]#  rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers  --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name          │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu        │ 0              │ 6                       │ 6        │ 1         │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1              │ 0                       │ 1        │ 0         │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘
[root@nullnull-os rabbitmq]#  rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers  --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name          │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu        │ 0              │ 1                       │ 1        │ 1         │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1              │ 0                       │ 1        │ 0         │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘
[root@nullnull-os rabbitmq]#  rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers  --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name          │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu        │ 0              │ 0                       │ 0        │ 1         │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1              │ 0                       │ 1        │ 0         │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘

使用拉确认消息

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.GetResponse;
import org.springframework.amqp.rabbit.core.ChannelCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.nio.charset.StandardCharsets;
import java.util.concurrent.ThreadLocalRandom;@RestController
public class MsgController {@Autowired private RabbitTemplate rabbitTemplate;@RequestMapping("/msg")public String getMessage() {String message =rabbitTemplate.execute(new ChannelCallback<String>() {@Overridepublic String doInRabbit(Channel channel) throws Exception {GetResponse getResponse = channel.basicGet("ack.qu", false);if (null == getResponse) {return "你已经消费完所有的消息";}String message = new String(getResponse.getBody(), StandardCharsets.UTF_8);if (ThreadLocalRandom.current().nextInt(10) % 3 == 0) {// 执行消息确认操作channel.basicAck(getResponse.getEnvelope().getDeliveryTag(), false);return "已确认的消息:" + message;} else {// 拒收一条消息并重新放回队列channel.basicReject(getResponse.getEnvelope().getDeliveryTag(), true);return "拒绝的消息:" + message;}}});return message;}
}

在浏览器中访问，同样有66%的概率会被拒绝，仅33%会被确认。

注：如果与监听在同一个工程，需将监听器给注释。

启动main函数，在浏览器中访问。http://127.0.0.1:8080/msg

可以看到返回:

拒绝的消息:消息:0
已确认的消息:消息:1
拒绝的消息:消息:2
......
已确认的消息:消息:9
你已经消费完所有的消息

同样的观察队列的一个消费情况：

[root@nullnull-os rabbitmq]#  rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers  --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name          │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu        │ 8              │ 0                       │ 8        │ 0         │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1              │ 0                       │ 1        │ 0         │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘
[root@nullnull-os rabbitmq]#  rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers  --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name          │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu        │ 3              │ 0                       │ 3        │ 0         │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1              │ 0                       │ 1        │ 0         │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘
[root@nullnull-os rabbitmq]#  rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers  --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name          │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu        │ 0              │ 0                       │ 0        │ 0         │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1              │ 0                       │ 1        │ 0         │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘
[root@nullnull-os rabbitmq]#

使用拉模式进行消息ACK确认也已经完成。

相关文章：

  • 四元数分析（Quaternion Analysis）在故障诊断中的应用
  • 手机录屏没有声音？让你的录屏有声有色
  • 鸿鹄电子招投标系统源码实现与立项流程：基于Spring Boot、Mybatis、Redis和Layui的企业电子招采平台
  • 【2023年中国高校大数据挑战赛 】赛题 B DNA 存储中的序列聚类与比对 Python实现
  • 详解ajax、fetch、axios的区别
  • 2023 CSIG青年科学家会议丨多模态大模型时代下的文档图像处理
  • C# windows服务程序开机自启动exe程序
  • CV之DL之Yolo：计算机视觉领域算法总结—Yolo系列(YoloV1~YoloV8各种对比)的简介、安装、案例应用之详细攻略
  • Linux之Shell编程
  • ubuntu apt 更换阿里云源
  • Linux常用命令大全＜二＞
  • windows监控进程是否还活着，查看内存使用率
  • Copilot在Pycharm的应用和示例
  • 关于目标检测中按照比例将数据集随机划分成训练集和测试集
  • 03 详细的Git命令使用大全
  • “Material Design”设计规范在 ComponentOne For WinForm 的全新尝试！
  • 0基础学习移动端适配
  • Hibernate【inverse和cascade属性】知识要点
  • Laravel Mix运行时关于es2015报错解决方案
  • PAT A1120
  • Python socket服务器端、客户端传送信息
  • Spring Boot MyBatis配置多种数据库
  • Spring Cloud Feign的两种使用姿势
  • vue-cli在webpack的配置文件探究
  • 猫头鹰的深夜翻译：JDK9 NotNullOrElse方法
  • 漂亮刷新控件-iOS
  • 通过获取异步加载JS文件进度实现一个canvas环形loading图
  • 为什么要用IPython/Jupyter?
  • 一天一个设计模式之JS实现——适配器模式
  • 原生JS动态加载JS、CSS文件及代码脚本
  • C# - 为值类型重定义相等性
  • Salesforce和SAP Netweaver里数据库表的元数据设计
  • ​HTTP与HTTPS：网络通信的安全卫士
  • ​力扣解法汇总1802. 有界数组中指定下标处的最大值
  • # Apache SeaTunnel 究竟是什么？
  • #微信小程序：微信小程序常见的配置传值
  • （2022版）一套教程搞定k8s安装到实战 | RBAC
  • （Redis使用系列） Springboot 实现Redis消息的订阅与分布 四
  • （大众金融）SQL server面试题（1）-总销售量最少的3个型号的车及其总销售量
  • （转）ObjectiveC 深浅拷贝学习
  • (转载)深入super，看Python如何解决钻石继承难题
  • (轉貼) UML中文FAQ (OO) (UML)
  • .gitignore
  • .NET Core 中插件式开发实现
  • .NET MVC之AOP
  • .NET 材料检测系统崩溃分析
  • .NET 同步与异步 之 原子操作和自旋锁（Interlocked、SpinLock）（九）
  • .netcore 如何获取系统中所有session_ASP.NET Core如何解决分布式Session一致性问题
  • @EventListener注解使用说明
  • [ C++ ] STL priority_queue(优先级队列)使用及其底层模拟实现，容器适配器，deque(双端队列)原理了解
  • [ vulhub漏洞复现篇 ] Django SQL注入漏洞复现 CVE-2021-35042
  • [145] 二叉树的后序遍历 js
  • [android] 手机卫士黑名单功能（ListView优化）
  • [ASP.NET MVC]如何定制Numeric属性/字段验证消息
  • [BZOJ 4598][Sdoi2016]模式字符串