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

    发送端确认机制

    ​ 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.rabbitmqgroupId>
                <artifactId>amqp-clientartifactId>
                <version>5.9.0version>
            dependency>

    • 1
    • 2
    • 3
    • 4
    • 5
    单条带确认模式
    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("等待消息确认超时");
    }
  }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47

    执行生产者的代码后，查看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]# 

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39

    经过检查发现消息成功成功的送到，并且收到了服务端的确认:

    发送的消息被确认:confirm 这是推送确认的消息

    • 1
    批量确认模式

    在以上的样例中，发送方在发送消息后，便进行入等待确认，这是一个同步阻塞的机制性能不是太好，接下来将使用批处理对此进行优化。

    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("等待消息确认超时");
    }
  }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62

    运行生产者，检查控制台输出:

    批量发送的消息被确认:
批量发送的消息被确认:
批量发送的消息被确认:
批量发送的消息被确认:
批量发送的消息被确认:
批量发送的消息被确认:
批量发送的消息被确认:
批量发送的消息被确认:
批量发送的消息被确认:
批量发送的消息被确认:
剩余发送的消息被确认:

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11

    可以观察到，数据已经成功的发送。

    再检查下队列的情况:

    [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]# 

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39

    检查服务端后发现，数据已经成功的发送至队列。

    异步回调模式

    除了批量确认外，还可以使用回调模式，回调模式，与批量相比，则是异步模式，不再会有阻塞的问题。

    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();
    }
  }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100
    • 101
    • 102
    • 103
    • 104
    • 105
    • 106
    • 107
    • 108
    • 109
    • 110
    • 111
    • 112
    • 113
    • 114
    • 115
    • 116
    • 117
    • 118
    • 119
    • 120
    • 121
    • 122
    • 123
    • 124
    • 125
    • 126
    • 127

    运行生产者的代码后，观察控制台：

    【发送】线程的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



    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55

    可以发现，生产者与确认机制，是完全异步的，生产者一个线程，而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]# 

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39

    至此使用异步监听回调模式已经完成。

    Spring发送端确认机制

    导入依赖

            <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-amqpartifactId>
        dependency>
        <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-webartifactId>
        dependency>

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    配制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

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18

    主入口类

    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);
  }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 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 {

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

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26

    控制类

    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";
  }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88

    运行应用程序,查看控制台输出:

    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


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65

    可以发现数据都已经成功的执行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]# 

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
  • 相关阅读:
    二分类问题的解决利器：逻辑回归算法详解（一）
    深入了解代理服务器：Socks5、IP代理与网络安全
    Java String.substring()方法具有什么功能呢？
    Vue项目流程7，交易页面，提交订单，支付页面，利用element UI 以及 QRCode 完成微信支付，弹出框按钮的相关工作，个人中心以及子路由我的订单
    ZooKeeper之Java API的基本使用以及应用场景的实现
    典型相关分析CCA计算过程
    iframe 实现跨域，两页面之间的通信
    SQL server 根据子级查询根父级
    以沙箱的方式运行容器:安全容器gvisor
    HDFS 分布式环境搭建
  • 原文地址：https://blog.csdn.net/bug_null/article/details/133249693