RabbitMQ消息可靠性保证机制--发送端确认

news2024/11/20 0:47:01

发送端确认机制

​ 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() {
            @Override
            public 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() {
            @Override
            public 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: publisherConfirm
  rabbitmq:
    host: node1
    virtual-host: /
    username: root
    password: 123456
    port: 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 {

  @Bean
  public Queue queue() {
    return new Queue("confirm.qe", false, false, false, null);
  }

  @Bean
  public Exchange exchange() {
    return new DirectExchange("confirm.ex", false, false, null);
  }

  @Bean
  public 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<>();

  @Autowired
  public void setTemplate(RabbitTemplate template) {
    this.template = template;

    // 设置回调函数
    RabbitTemplate.ConfirmCallback ackCallback =
        new RabbitTemplate.ConfirmCallback() {
          @Override
          public 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]#

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1046458.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

python使用uiautomator2操作真机

python使用uiautomator2操作真机

测试环境&#xff1a;win10 64位&#xff0c;python3.10.4&#xff1b;真机&#xff0c;荣耀10青春版&#xff0c;Android版本10。 之前是在手机模拟器上操作的&#xff0c;参考我的文章python使用uiautomator2操作雷电模拟器_小小爬虾的博客-CSDN博客 一、将手机设置为开发者…
阅读更多...
任务执行大数据量与高并发方案

任务执行大数据量与高并发方案

大数据量高并发任务解决方案 场景 每个任务有十万条以上的数据&#xff0c;任务执行过程中对这些数据逐条做分析处理。 在同一段时间&#xff0c;会出现任务高并发执行&#xff0c;导致内存溢出 解决方案 1、分批处理 任务执行过程中&#xff0c;不一次性读取全量数据&…
阅读更多...
将切分的图片筛选出有缺陷的

将切分的图片筛选出有缺陷的

将切分的图片筛选出有缺陷的 需求代码 需求 由于之前切分的图像有一些存在没有缺陷&#xff0c;需要再次筛选 将可视化的图像更改后缀 更改为xml的 可视化代码 可视化后只有7000多个图像 原本的图像有1W多张 代码 # 按照xml文件删除对应的图片 # coding: utf-8 from P…
阅读更多...
Java比较器之equals、comparable、comparator

Java比较器之equals、comparable、comparator

文章目录 前言一、基本类型比较1.2.equals3.和equals的区别 二、对象的比较1.覆写基类的equals2.基于Comparable接口类的比较3.基于Comparator比较器比较4.三种方式对比 前言 在Java中&#xff0c;基本类型的对象可以直接比较&#xff0c;而自定义类型&#xff0c;默认是用equ…
阅读更多...
秋招面经记录

秋招面经记录

秋招面经记录 MySQLRedis项目分布式框架java网络数据结构设计模式HR手撕 MySQL Mysql中有1000万条数据&#xff0c;每次查询10条&#xff0c;该如何优化&#xff08;答&#xff1a;Limit子查询优化&#xff09; select t.* from t_topic t LIMIT 90000,10; 对上面的mysql语句说…
阅读更多...
【Java 进阶篇】数据定义语言（DDL）详解

【Java 进阶篇】数据定义语言（DDL）详解

数据定义语言&#xff08;DDL&#xff09;是SQL&#xff08;结构化查询语言&#xff09;的一部分&#xff0c;它用于定义、管理和控制数据库的结构和元素。DDL允许数据库管理员、开发人员和其他用户创建、修改和删除数据库对象&#xff0c;如表、索引、视图等。在本文中&#x…
阅读更多...
【C++杂货店】类和对象（上）

【C++杂货店】类和对象（上）

【C杂货店】类和对象&#xff08;上&#xff09; 一、面向过程和面向对象初步认识二、类的引入三、类的定义四、类的访问限定符及封装4.1 访问限定符4.2 封装 五、类的作用域六、类的实例化七、类对象模型7.1 类对象的存储规则7.2 例题7.3结构体内存对齐规则 八、this指针8.2 t…
阅读更多...
【Spring Cloud】认识微服务架构，拆分简单的 Demo 实现服务的远程调用

【Spring Cloud】认识微服务架构，拆分简单的 Demo 实现服务的远程调用

文章目录 前言一、认识微服务1.1 服务架构的演变&#xff1a;从单体到微服务单体架构分布式架构微服务架构 1.2 微服务技术的对比&#xff1a;Dubbo、Spring Cloud、Spring Cloud Alibaba技术对比公司需求的选择 1.3 Spring Cloud&#xff1a;微服务框架的精华什么是 Spring Cl…
阅读更多...
【Vue.js】使用ElementUI实现增删改查(CRUD)及表单验证

【Vue.js】使用ElementUI实现增删改查(CRUD)及表单验证

前言&#xff1a; 本文根据上篇实现数据表格(查所有)完善增删改功能&#xff0c;数据表格》查看数据表格的实现链接 一&#xff0c;增删改查 ①后端Controller(增删改查方法)&#xff1a; package com.zking.ssm.controller;import com.zking.ssm.model.Book; import com.z…
阅读更多...
新能源汽车行业出口ERP管理解决方案

新能源汽车行业出口ERP管理解决方案

中国汽车企业以史无前例的规模进军慕尼黑车展。本届展会&#xff0c;中国汽车参展企业数量达50家&#xff0c;是2021年的两倍。欧洲销售的新型电动汽车中&#xff0c;8%由中国品牌制造。2022年上半年&#xff0c;中国电动汽车的平均价格不到3.2万欧元&#xff08;3.5万美元&…
阅读更多...
香港云服务器和日本云服务器哪个好？（详细对比）

香港云服务器和日本云服务器哪个好？（详细对比）

​　　购置海外服务器时&#xff0c;您是在乎网络速度?价格?稳定性?当这几个因素同时存在&#xff0c;我们该如何选择?本篇针对海外热门的两个地区&#xff0c;中国香港和日本&#xff0c;这两种云服务器谁优谁劣?各有什么亮点?逐一进行对比分析。 一、速度上来看 中国香…
阅读更多...
Tungsten Fabric数据量过大问题处理初探

Tungsten Fabric数据量过大问题处理初探

开源SDN系统Tungsten Fabric面临数据产生过多问题。 经排查&#xff0c;产生数据多出自analytics组件的Cassandra数据库()。很多分析数据会存储至Cassandra库&#xff0c;并持久化处理。 没有特殊调整的话&#xff0c;目录在 /var/lib/docker/volumes/analytics_database_an…
阅读更多...
春招秋招，大学生求职容易遇到哪些问题？

春招秋招，大学生求职容易遇到哪些问题？

每到毕业季就有大批大学生从校园出来&#xff0c;他们怀抱梦想&#xff0c;希望能做出一番成绩。但现实总归是残酷的&#xff0c;有些人找不到工作&#xff0c;有一些人频繁跳槽&#xff0c;也有一些人最终找到的工作与自己的专业没有一点关系&#xff0c;迷茫好几年才找到方向…
阅读更多...
数字音频工作站FL Studio 21中文版下载及电音编曲要用乐理吗 电音编曲步骤

数字音频工作站FL Studio 21中文版下载及电音编曲要用乐理吗 电音编曲步骤

FL Studio 21是一款强大的数字音频工作站&#xff08;DAW&#xff09;软件&#xff0c;为您提供一个完整的软件音乐制作环境。它是制作高质量的音乐、乐器、录音等的完整解决方案。该程序配备了各种工具和插件&#xff0c;帮助你创建专业的虚拟乐器&#xff0c;如贝斯、吉他、钢…
阅读更多...
K8S笔记

K8S笔记

阅读更多...
探索Moonbeam路由流动性的强大功能

探索Moonbeam路由流动性的强大功能

Moonbeam的GMP预编译作为MRL的接口&#xff0c;有助于将带有Token的消息从GMP协议&#xff08;通过XCMP&#xff09;传输到与Moonbeam链接的平行链。 为何是个重磅消息&#xff1f;因为这项技术使得将流动性从外部区块链转移到其他波卡平行链成为可能&#xff01; 这里补充一…
阅读更多...
快手商品详情数据接口

快手商品详情数据接口

快手商品详情数据接口的调用需要使用快手提供的API接口。API接口是一种程序之间进行交互的方式&#xff0c;它允许两个程序或者网页之间互相通信、交换数据。 快手API接口的具体使用方法和步骤需要您参考快手的官方文档或者开发者指南。在使用快手API接口时&#xff0c;您需要…
阅读更多...
基于遗传算法解决的多仓库多旅行推销员问题（Matlab代码实现）

基于遗传算法解决的多仓库多旅行推销员问题（Matlab代码实现）

&#x1f4a5;&#x1f4a5;&#x1f49e;&#x1f49e;欢迎来到本博客❤️❤️&#x1f4a5;&#x1f4a5; &#x1f3c6;博主优势&#xff1a;&#x1f31e;&#x1f31e;&#x1f31e;博客内容尽量做到思维缜密&#xff0c;逻辑清晰&#xff0c;为了方便读者。 ⛳️座右铭&a…
阅读更多...
【Python基础】if __name__ == ‘__main__‘:和assert函数

【Python基础】if __name__ == ‘__main__‘:和assert函数

&#x1f4e2;&#xff1a;如果你也对机器人、人工智能感兴趣&#xff0c;看来我们志同道合✨ &#x1f4e2;&#xff1a;不妨浏览一下我的博客主页【https://blog.csdn.net/weixin_51244852】 &#x1f4e2;&#xff1a;文章若有幸对你有帮助&#xff0c;可点赞 &#x1f44d;…
阅读更多...
day49 jdbc技术

day49 jdbc技术

一、概述 什么是JDBC Java DataBase COnnectivity Java 数据库连接 其实就是利用Java程序连接并访问数据库的一种技术 为什么要学习JDBC 之前我们是通过终端&#xff0c;或者第三方工具直接连接数据库 在企业开发中&#xff0c;更多的是通过程序来连接数据库的 未来学习的M…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023