发送端确认机制
RabbitMQ后来引入了一种轻量级的方式，叫发送方确认(publisher confirm)机制，生产者将信息设置成confirm（确认）模式，一旦信道进入了confirm模式，所有在该信道上面发送的消息都会被指派成一个唯一的ID（从1开始），一旦消息被投递到所有匹配的队列之后（如果消息和队列是持久化的，那么消息会在消息持久化后发出），RabbitMQ就会发送一个确认（Basic.Ack）给生产者（包含消息的唯一的ID），这样生产者就知道消息已经正确送达了。
RabbitMQ回传给生产者的确认消息中的devliveryTag字段包含了确认消息的序号，另外通过设置channel.basicAck方法中的mulitiple参数，表示到这个序号之前的所有消息是否都已经得到了处理了。生产都投递消息后并不需要一直阻塞着，可以继续投递下一条消息并通过回调方式处理ACK响应。如果RabbitMQ因数自身内部错误导致消息丢失等异常情况发生，就会响应一条nack（Basic.Nack)命令，生产者应用程序同样可以在回调方法中处理该nack命令。
首先导入maven依赖
<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.9.0</version></dependency>
单条带确认模式
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;public class Product {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();// 开启发送方确认机制AMQP.Confirm.SelectOk selectOk = channel.confirmSelect();channel.exchangeDeclare("confirm.ex", BuiltinExchangeType.DIRECT);channel.queueDeclare("confirm.qe", false, false, false, null);channel.queueBind("confirm.qe", "confirm.ex", "confirm.rk");String pushMsg = "confirm 这是推送确认的消息";channel.basicPublish("confirm.ex", "confirm.rk", null, pushMsg.getBytes(StandardCharsets.UTF_8));// 执行发送端确认机制try {channel.waitForConfirmsOrDie(5000);System.out.println("发送的消息被确认:" + pushMsg);} catch (IOException e) {e.printStackTrace();System.out.println("消息被拒绝:" + pushMsg);} catch (InterruptedException e) {e.printStackTrace();System.out.println("非Publisher confirm的通道上使用该方法");} catch (TimeoutException e) {e.printStackTrace();System.out.println("等待消息确认超时");}}
}
执行生产者的代码后，查看RabbitMQ中的是否到达:
[root@nullnull-os rabbitmq]# rabbitmqctl list_exchanges --formatter pretty_table
Listing exchanges for vhost / ...
┌────────────────────┬─────────┐
│ name │ type │
├────────────────────┼─────────┤
│ amq.fanout │ fanout │
├────────────────────┼─────────┤
│ confirm.ex │ direct │
├────────────────────┼─────────┤
│ amq.rabbitmq.trace │ topic │
├────────────────────┼─────────┤
│ amq.headers │ headers │
├────────────────────┼─────────┤
│ amq.topic │ topic │
├────────────────────┼─────────┤
│ amq.direct │ direct │
├────────────────────┼─────────┤
│ │ direct │
├────────────────────┼─────────┤
│ amq.match │ headers │
└────────────────────┴─────────┘
[root@nullnull-os rabbitmq]# rabbitmqctl list_bindings --formatter pretty_table
Listing bindings for vhost /...
┌─────────────┬─────────────┬──────────────────┬──────────────────┬─────────────┬───────────┐
│ source_name │ source_kind │ destination_name │ destination_kind │ routing_key │ arguments │
├─────────────┼─────────────┼──────────────────┼──────────────────┼─────────────┼───────────┤
│ │ exchange │ confirm.qe │ queue │ confirm.qe │ │
├─────────────┼─────────────┼──────────────────┼──────────────────┼─────────────┼───────────┤
│ confirm.ex │ exchange │ confirm.qe │ queue │ confirm.rk │ │
└─────────────┴─────────────┴──────────────────┴──────────────────┴─────────────┴───────────┘
[root@nullnull-os rabbitmq]# rabbitmqctl list_queues --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌────────────┬──────────┐
│ name │ messages │
├────────────┼──────────┤
│ confirm.qe │ 1 │
└────────────┴──────────┘
[root@nullnull-os rabbitmq]#
经过检查发现消息成功成功的送到，并且收到了服务端的确认:
发送的消息被确认:confirm 这是推送确认的消息
批量确认模式
在以上的样例中，发送方在发送消息后，便进行入等待确认，这是一个同步阻塞的机制性能不是太好，接下来将使用批处理对此进行优化。
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;public class ProductBatch {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();// 开启发送方确认机制AMQP.Confirm.SelectOk selectOk = channel.confirmSelect();channel.exchangeDeclare("confirm.ex", BuiltinExchangeType.DIRECT);channel.queueDeclare("confirm.qe", false, false, false, null);channel.queueBind("confirm.qe", "confirm.ex", "confirm.rk");// 批量处理大小，即每10条进行一次确认int batch = 10;// 确认的计数int confirmNum = 0;// 执行发送端确认机制try {for (int i = 0; i < 108; i++) {confirmNum++;String pushMsg = "confirm 这是推送确认的消息" + i;channel.basicPublish("confirm.ex", "confirm.rk", null, pushMsg.getBytes(StandardCharsets.UTF_8));if (confirmNum == batch) {channel.waitForConfirmsOrDie(5000);System.out.println("批量发送的消息被确认:");confirmNum = 0;}}if (confirmNum > 0) {channel.waitForConfirmsOrDie(5000);System.out.println("剩余发送的消息被确认:");}} catch (IOException e) {e.printStackTrace();System.out.println("消息被拒绝:");} catch (InterruptedException e) {e.printStackTrace();System.out.println("非Publisher confirm的通道上使用该方法");} catch (TimeoutException e) {e.printStackTrace();System.out.println("等待消息确认超时");}}
}运行生产者，检查控制台输出:
批量发送的消息被确认:
批量发送的消息被确认:
批量发送的消息被确认:
批量发送的消息被确认:
批量发送的消息被确认:
批量发送的消息被确认:
批量发送的消息被确认:
批量发送的消息被确认:
批量发送的消息被确认:
批量发送的消息被确认:
剩余发送的消息被确认:
可以观察到，数据已经成功的发送。
再检查下队列的情况:
[root@nullnull-os rabbitmq]# rabbitmqctl list_exchanges --formatter pretty_table
Listing exchanges for vhost / ...
┌────────────────────┬─────────┐
│ name │ type │
├────────────────────┼─────────┤
│ amq.fanout │ fanout │
├────────────────────┼─────────┤
│ confirm.ex │ direct │
├────────────────────┼─────────┤
│ amq.rabbitmq.trace │ topic │
├────────────────────┼─────────┤
│ amq.headers │ headers │
├────────────────────┼─────────┤
│ amq.topic │ topic │
├────────────────────┼─────────┤
│ amq.direct │ direct │
├────────────────────┼─────────┤
│ │ direct │
├────────────────────┼─────────┤
│ amq.match │ headers │
└────────────────────┴─────────┘
[root@nullnull-os rabbitmq]# rabbitmqctl list_bindings --formatter pretty_table
Listing bindings for vhost /...
┌─────────────┬─────────────┬──────────────────┬──────────────────┬─────────────┬───────────┐
│ source_name │ source_kind │ destination_name │ destination_kind │ routing_key │ arguments │
├─────────────┼─────────────┼──────────────────┼──────────────────┼─────────────┼───────────┤
│ │ exchange │ confirm.qe │ queue │ confirm.qe │ │
├─────────────┼─────────────┼──────────────────┼──────────────────┼─────────────┼───────────┤
│ confirm.ex │ exchange │ confirm.qe │ queue │ confirm.rk │ │
└─────────────┴─────────────┴──────────────────┴──────────────────┴─────────────┴───────────┘
[root@nullnull-os rabbitmq]# rabbitmqctl list_queues --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌────────────┬──────────┐
│ name │ messages │
├────────────┼──────────┤
│ confirm.qe │ 108 │
└────────────┴──────────┘
[root@nullnull-os rabbitmq]#
检查服务端后发现，数据已经成功的发送至队列。
异步回调模式
除了批量确认外，还可以使用回调模式，回调模式，与批量相比，则是异步模式，不再会有阻塞的问题。
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeoutException;public class ProductCallBack {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();try {// 开启发送方确认机制AMQP.Confirm.SelectOk selectOk = channel.confirmSelect();channel.exchangeDeclare("confirm.ex", BuiltinExchangeType.DIRECT);channel.queueDeclare("confirm.qe", false, false, false, null);channel.queueBind("confirm.qe", "confirm.ex", "confirm.rk");// 记录下发送与确认的消息ConcurrentNavigableMap<Long, String> ackConfirmMap = new ConcurrentSkipListMap<>();ConcurrentNavigableMap<Long, String> nackConfirmMap = new ConcurrentSkipListMap<>();// ack确认的信息ConfirmCallback ackCallback =new ConfirmCallback() {@Overridepublic void handle(long deliveryTag, boolean multiple) throws IOException {if (multiple) {// 获取已经被确认的Map集合ConcurrentNavigableMap<Long, String> headMap =ackConfirmMap.headMap(deliveryTag, true);long threadId = Thread.currentThread().getId();System.out.println("【ack确认】线程ID："+ threadId+ "，批量：小于等于:"+ deliveryTag+ "的消息都被确认了，内容："+ headMap.keySet());headMap.clear();} else {long threadId = Thread.currentThread().getId();ackConfirmMap.remove(deliveryTag);System.out.println("【ack确认】线程ID：" + threadId + "，单条:" + deliveryTag + "对应的消息被确认,key:" + deliveryTag);}}};// 不确认的消息ConfirmCallback nackCallBack =new ConfirmCallback() {@Overridepublic void handle(long deliveryTag, boolean multiple) throws IOException {if (multiple) {ConcurrentNavigableMap<Long, String> headMap =ackConfirmMap.headMap(deliveryTag, true);long threadId = Thread.currentThread().getId();System.out.println("【nack确认】线程ID："+ threadId+ "，批量：小于等于"+ deliveryTag+ "的消息都不被确认了,内容:"+ headMap.keySet());nackConfirmMap.putAll(headMap);headMap.clear();} else {long threadId = Thread.currentThread().getId();System.out.println("【nack确认】线程的ID：" + threadId + "，单条:" + deliveryTag + "对应的消息被确认");String value = ackConfirmMap.remove(deliveryTag);nackConfirmMap.put(deliveryTag, value);}}};channel.addConfirmListener(ackCallback, nackCallBack);// 执行数据发送try {for (int i = 0; i < 28; i++) {// 获取当前发送的序列号long nextPublishSeqNo = channel.getNextPublishSeqNo();String pushMsg = "序列号:" + nextPublishSeqNo + ":已经发送了消息信息:" + (i + 1);channel.basicPublish("confirm.ex", "confirm.rk", null, pushMsg.getBytes(StandardCharsets.UTF_8));ackConfirmMap.put(nextPublishSeqNo, pushMsg);long threadId = Thread.currentThread().getId();System.out.println("【发送】线程的ID：" + threadId + "，序号：" + nextPublishSeqNo + "发送完毕");// 随机休眠Thread.sleep(ThreadLocalRandom.current().nextInt(0, 5));}} catch (IOException e) {e.printStackTrace();System.out.println("消息被拒绝:");}Thread.sleep(5000);System.out.println("确认的消息:" + ackConfirmMap.size());System.out.println("未确认的消息:" + nackConfirmMap.size());} catch (IOException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();} finally {channel.close();connection.close();}}
}运行生产者的代码后，观察控制台：
【发送】线程的ID：1，序号：1发送完毕
【发送】线程的ID：1，序号：2发送完毕
【发送】线程的ID：1，序号：3发送完毕
【ack确认】线程ID：20，单条:1对应的消息被确认,key:1
【发送】线程的ID：1，序号：4发送完毕
【发送】线程的ID：1，序号：5发送完毕
【发送】线程的ID：1，序号：6发送完毕
【ack确认】线程ID：20，单条:2对应的消息被确认,key:2
【ack确认】线程ID：20，单条:3对应的消息被确认,key:3
【发送】线程的ID：1，序号：7发送完毕
【发送】线程的ID：1，序号：8发送完毕
【发送】线程的ID：1，序号：9发送完毕
【ack确认】线程ID：20，批量：小于等于:6的消息都被确认了，内容：[4, 5, 6]
【发送】线程的ID：1，序号：10发送完毕
【发送】线程的ID：1，序号：11发送完毕
【ack确认】线程ID：20，批量：小于等于:8的消息都被确认了，内容：[7, 8]
【ack确认】线程ID：20，单条:9对应的消息被确认,key:9
【发送】线程的ID：1，序号：12发送完毕
【发送】线程的ID：1，序号：13发送完毕
【ack确认】线程ID：20，单条:10对应的消息被确认,key:10
【ack确认】线程ID：20，单条:11对应的消息被确认,key:11
【发送】线程的ID：1，序号：14发送完毕
【ack确认】线程ID：20，单条:12对应的消息被确认,key:12
【发送】线程的ID：1，序号：15发送完毕
【发送】线程的ID：1，序号：16发送完毕
【发送】线程的ID：1，序号：17发送完毕
【ack确认】线程ID：20，单条:13对应的消息被确认,key:13
【ack确认】线程ID：20，单条:14对应的消息被确认,key:14
【发送】线程的ID：1，序号：18发送完毕
【ack确认】线程ID：20，单条:15对应的消息被确认,key:15
【发送】线程的ID：1，序号：19发送完毕
【ack确认】线程ID：20，单条:16对应的消息被确认,key:16
【ack确认】线程ID：20，单条:17对应的消息被确认,key:17
【发送】线程的ID：1，序号：20发送完毕
【发送】线程的ID：1，序号：21发送完毕
【ack确认】线程ID：20，单条:18对应的消息被确认,key:18
【发送】线程的ID：1，序号：22发送完毕
【发送】线程的ID：1，序号：23发送完毕
【ack确认】线程ID：20，单条:19对应的消息被确认,key:19
【发送】线程的ID：1，序号：24发送完毕
【ack确认】线程ID：20，单条:20对应的消息被确认,key:20
【ack确认】线程ID：20，单条:21对应的消息被确认,key:21
【发送】线程的ID：1，序号：25发送完毕
【发送】线程的ID：1，序号：26发送完毕
【发送】线程的ID：1，序号：27发送完毕
【发送】线程的ID：1，序号：28发送完毕
【ack确认】线程ID：20，批量：小于等于:23的消息都被确认了，内容：[22, 23]
【ack确认】线程ID：20，单条:24对应的消息被确认,key:24
【ack确认】线程ID：20，批量：小于等于:26的消息都被确认了，内容：[25, 26]
【ack确认】线程ID：20，单条:27对应的消息被确认,key:27
【ack确认】线程ID：20，单条:28对应的消息被确认,key:28
确认的消息:0
未确认的消息:0可以发现，生产者与确认机制，是完全异步的，生产者一个线程，而ack在另外的一个线程。可以并行处理。
最后再来检查下队列的信息：
[root@nullnull-os rabbitmq]# rabbitmqctl list_exchanges --formatter pretty_table
Listing exchanges for vhost / ...
┌────────────────────┬─────────┐
│ name │ type │
├────────────────────┼─────────┤
│ amq.fanout │ fanout │
├────────────────────┼─────────┤
│ confirm.ex │ direct │
├────────────────────┼─────────┤
│ amq.rabbitmq.trace │ topic │
├────────────────────┼─────────┤
│ amq.headers │ headers │
├────────────────────┼─────────┤
│ amq.topic │ topic │
├────────────────────┼─────────┤
│ amq.direct │ direct │
├────────────────────┼─────────┤
│ │ direct │
├────────────────────┼─────────┤
│ amq.match │ headers │
└────────────────────┴─────────┘
[root@nullnull-os rabbitmq]# rabbitmqctl list_bindings --formatter pretty_table
Listing bindings for vhost /...
┌─────────────┬─────────────┬──────────────────┬──────────────────┬─────────────┬───────────┐
│ source_name │ source_kind │ destination_name │ destination_kind │ routing_key │ arguments │
├─────────────┼─────────────┼──────────────────┼──────────────────┼─────────────┼───────────┤
│ │ exchange │ confirm.qe │ queue │ confirm.qe │ │
├─────────────┼─────────────┼──────────────────┼──────────────────┼─────────────┼───────────┤
│ confirm.ex │ exchange │ confirm.qe │ queue │ confirm.rk │ │
└─────────────┴─────────────┴──────────────────┴──────────────────┴─────────────┴───────────┘
[root@nullnull-os rabbitmq]# rabbitmqctl list_queues --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌────────────┬──────────┐
│ name │ messages │
├────────────┼──────────┤
│ confirm.qe │ 28 │
└────────────┴──────────┘
[root@nullnull-os rabbitmq]#
至此使用异步监听回调模式已经完成。
Spring发送端确认机制
导入依赖
<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>
配制RabbitMQ
spring:application:name: publisherConfirmrabbitmq:host: node1virtual-host: /username: rootpassword: 123456port: 5672# 启用发送方确认机制publisher-returns: true#NONE值是禁用发布确认模式，是默认值#CORRELATED值是发布消息成功到交换器后会触发回调方法#SIMPLE值经测试有两种效果，其一效果和CORRELATED值一样会触发回调方法，#其二在发布消息成功后使用rabbitTemplate调用waitForConfirms或waitForConfirmsOrDie#方法等待broker节点返回发送结果，根据返回结果来判定下一步的逻辑，#要注意的点是waitForConfirmsOrDie方法如果返回false则会关闭channel，则接下来无法发送消息到broker;publisher-confirm-type: correlated
主入口类
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class PublisherConfirmApplication {public static void main(String[] args) {SpringApplication.run(PublisherConfirmApplication.class, args);}
}
队列配制
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("confirm.qe", false, false, false, null);}@Beanpublic Exchange exchange() {return new DirectExchange("confirm.ex", false, false, null);}@Beanpublic Binding binding() {return BindingBuilder.bind(queue()).to(exchange()).with("confirm.rk").noargs();}
}
控制类
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.MessagePropertiesBuilder;
import org.springframework.amqp.rabbit.connection.CorrelationData;
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.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ThreadLocalRandom;@RestController
public class DataController {private RabbitTemplate template;/** 用于记录下发送的容器 */private Map<Long, String> ackMap = new ConcurrentHashMap<>();@Autowiredpublic void setTemplate(RabbitTemplate template) {this.template = template;// 设置回调函数RabbitTemplate.ConfirmCallback ackCallback =new RabbitTemplate.ConfirmCallback() {@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {if (ack) {String msg =new String(correlationData.getReturnedMessage().getBody(), StandardCharsets.UTF_8);ackMap.remove(Long.parseLong(correlationData.getId()));System.out.println("【回调】消息确认：" + correlationData.getId() + "--" + msg);} else {System.out.println("异常：" + cause);}}};this.template.setConfirmCallback(ackCallback);}@RequestMapping("/biz")public String doInvoke() throws Exception {long sendSeq = 1;for (int i = 0; i < 28; i++) {MessageProperties build =MessagePropertiesBuilder.newInstance().setHeader("key1", "value1").setCorrelationId(sendSeq + "").setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN).build();build.setConsumerTag("msg");CorrelationData dataSend = new CorrelationData();dataSend.setId(sendSeq + "");// 用于回调验证的消息byte[] returnBytes = "这是响应的消息:".getBytes(StandardCharsets.UTF_8);dataSend.setReturnedMessage(new Message(returnBytes, null));// 发送的消息String msg = "这是等待确认的消息";byte[] sendBytes = msg.getBytes(StandardCharsets.UTF_8);Message message = new Message(sendBytes, build);template.convertAndSend("confirm.ex", "confirm.rk", message, dataSend);ackMap.put(sendSeq, msg);System.out.println("【发送】发送成功:" + sendSeq);Thread.sleep(ThreadLocalRandom.current().nextInt(0, 10));sendSeq = sendSeq + 1;}Thread.sleep(3000);System.out.println("未确认ACK的消息:" + ackMap.size());return "ok";}
}
运行应用程序,查看控制台输出:
2023-08-21 22:33:24.969 INFO 8628 --- [nio-8080-exec-1] o.a.c.c.C.[Tomcat].[localhost].[/] : Initializing Spring DispatcherServlet 'dispatcherServlet'
2023-08-21 22:33:24.969 INFO 8628 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet : Initializing Servlet 'dispatcherServlet'
2023-08-21 22:33:24.974 INFO 8628 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet : Completed initialization in 5 ms
2023-08-21 22:33:25.001 INFO 8628 --- [nio-8080-exec-1] o.s.a.r.c.CachingConnectionFactory : Attempting to connect to: [node1:5672]
2023-08-21 22:33:25.119 INFO 8628 --- [nio-8080-exec-1] o.s.a.r.c.CachingConnectionFactory : Created new connection: rabbitConnectionFactory#3bd6ba24:0/SimpleConnection@5b31efba [delegate=amqp://root@150.158.137.207:5672/, localPort= 63833]
2023-08-21 22:33:25.123 INFO 8628 --- [nio-8080-exec-1] o.s.amqp.rabbit.core.RabbitAdmin : Auto-declaring a non-durable or auto-delete Exchange (confirm.ex) durable:false, auto-delete:false. It will be deleted by the broker if it shuts down, and can be redeclared by closing and reopening the connection.
2023-08-21 22:33:25.123 INFO 8628 --- [nio-8080-exec-1] o.s.amqp.rabbit.core.RabbitAdmin : Auto-declaring a non-durable, auto-delete, or exclusive Queue (confirm.qe) durable:false, auto-delete:false, exclusive:false. It will be redeclared if the broker stops and is restarted while the connection factory is alive, but all messages will be lost.
【发送】发送成功:1
【回调】消息确认：1--这是响应的消息:
【发送】发送成功:2
【发送】发送成功:3
【回调】消息确认：2--这是响应的消息:
【回调】消息确认：3--这是响应的消息:
【发送】发送成功:4
【发送】发送成功:5
【发送】发送成功:6
【回调】消息确认：4--这是响应的消息:
【发送】发送成功:7
【回调】消息确认：5--这是响应的消息:
【回调】消息确认：6--这是响应的消息:
【回调】消息确认：7--这是响应的消息:
【发送】发送成功:8
【发送】发送成功:9
【发送】发送成功:10
【发送】发送成功:11
【回调】消息确认：8--这是响应的消息:
【回调】消息确认：9--这是响应的消息:
【回调】消息确认：10--这是响应的消息:
【回调】消息确认：11--这是响应的消息:
【发送】发送成功:12
【发送】发送成功:13
【发送】发送成功:14
【回调】消息确认：12--这是响应的消息:
【发送】发送成功:15
【发送】发送成功:16
【回调】消息确认：13--这是响应的消息:
【发送】发送成功:17
【回调】消息确认：14--这是响应的消息:
【发送】发送成功:18
【发送】发送成功:19
【回调】消息确认：15--这是响应的消息:
【发送】发送成功:20
【回调】消息确认：16--这是响应的消息:
【回调】消息确认：17--这是响应的消息:
【回调】消息确认：18--这是响应的消息:
【回调】消息确认：19--这是响应的消息:
【回调】消息确认：20--这是响应的消息:
【发送】发送成功:21
【发送】发送成功:22
【发送】发送成功:23
【回调】消息确认：21--这是响应的消息:
【回调】消息确认：22--这是响应的消息:
【发送】发送成功:24
【回调】消息确认：23--这是响应的消息:
【发送】发送成功:25
【回调】消息确认：24--这是响应的消息:
【发送】发送成功:26
【回调】消息确认：25--这是响应的消息:
【发送】发送成功:27
【发送】发送成功:28
【回调】消息确认：26--这是响应的消息:
【回调】消息确认：27--这是响应的消息:
【回调】消息确认：28--这是响应的消息:
未确认ACK的消息:0可以发现数据都已经成功的执行ACK的确认机制。此与原生的API，还是存在一些不同。此在发送时，已经将响应关联的ID进行了指定。这样当收到了confirm时，即能与之前发送的数据关联上。
检查队列的信息：
[root@nullnull-os rabbitmq]# rabbitmqctl list_exchanges --formatter pretty_table
Listing exchanges for vhost / ...
┌────────────────────┬─────────┐
│ name │ type │
├────────────────────┼─────────┤
│ amq.fanout │ fanout │
├────────────────────┼─────────┤
│ confirm.ex │ direct │
├────────────────────┼─────────┤
│ amq.rabbitmq.trace │ topic │
├────────────────────┼─────────┤
│ amq.headers │ headers │
├────────────────────┼─────────┤
│ amq.topic │ topic │
├────────────────────┼─────────┤
│ amq.direct │ direct │
├────────────────────┼─────────┤
│ │ direct │
├────────────────────┼─────────┤
│ amq.match │ headers │
└────────────────────┴─────────┘
[root@nullnull-os rabbitmq]# rabbitmqctl list_bindings --formatter pretty_table
Listing bindings for vhost /...
┌─────────────┬─────────────┬──────────────────┬──────────────────┬─────────────┬───────────┐
│ source_name │ source_kind │ destination_name │ destination_kind │ routing_key │ arguments │
├─────────────┼─────────────┼──────────────────┼──────────────────┼─────────────┼───────────┤
│ │ exchange │ confirm.qe │ queue │ confirm.qe │ │
├─────────────┼─────────────┼──────────────────┼──────────────────┼─────────────┼───────────┤
│ confirm.ex │ exchange │ confirm.qe │ queue │ confirm.rk │ │
└─────────────┴─────────────┴──────────────────┴──────────────────┴─────────────┴───────────┘
[root@nullnull-os rabbitmq]# rabbitmqctl list_queues --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌────────────┬──────────┐
│ name │ messages │
├────────────┼──────────┤
│ confirm.qe │ 28 │
└────────────┴──────────┘
[root@nullnull-os rabbitmq]#
