【RabbitMQ】——消息应答

<1> 概念

消费者完成一个任务可能需要一段时间，如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了，会发生什么情况。RabbitMQ一旦向消费者传递了一条消息，便立即将该消息标记为删除。在这种情况下，突然有个消费者挂掉了，我们将丢失正在处理的消息。以及后续发送给该消费这的消息，因为它无法接收到。
为了保证消息在发送过程中不丢失，rabbitmq_引入消息应答机制，消息应答就是:消费者在接收到消息并且处理该消息之后，告诉rabbitmq它已经处理了，rabbitmq可以把该消息删除了。

<2> 自动应答

消息发送后立即被认为已经传送成功，这种模式需要在高吞吐量和数据传输安全性方面做权衡,因为这种模式如果消息在接收到之前，消费者那边出现连接或者channel 关闭，那么消息就丢失了,当然另一方面这种模式消费者那边可以传递过载的消息，没有对传递的消息数量进行限制，当然这样有可能使得消费者这边由于接收太多还来不及处理的消息，导致这些消息的积压，最终使得内存耗尽，最终这些消费者线程被操作系统杀死，所以这种模式仅适用在消费者可以高效并以某种速率能够处理这些消息的情况下使用。

<3>手动应答

1. Channel.basicAck()

用于肯定确认
RabbitMQ已经知道该消息并且处理成功，可以将其丢弃了

2. Channel.basicNack()

用于否定确认

3. Channel.basicReject()

用于否定确认
比Channel. basicNack()少了一个批量处理的参数
建议不要批量处理
不处理该消息了直接拒接。可以将其丢弃了

4. 代码实现手动应答

! 注意，在之前的博客中给过的工具类，在本文中不在给出，详情请翻阅之前的博客！

生产者 TaskProducerAck

package com.rabbitmqDemo.rabbitmq.three;

import com.rabbitmq.client.Channel;
import com.rabbitmqDemo.rabbitmq.utils.RabbitMqUtils;

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

/**
 * 生产者, 消息在手动应答时不丢失，放回队列重新消费
 */
public class TaskProducerAck {

    //队列名称
    private static final String QUEUE_NAME = "ack_queue";

    /**
     * 发送消息
     */
    public static void main(String[] args) throws IOException, TimeoutException {
        //获取信道
        Channel channel = RabbitMqUtils.getChannel();
        /**
         * 生成一个队列
         * param1 队列名字
         * param2 队列里的消息是否持久化（磁盘），默认情况消息存储在内存
         * param3 该队列是否只供一个消费者进行消费，是否进行消息共享，true可以多个消费者消费，false 不允许多个消费者消费
         * param4 是否自动删除，最后一个消费断开连接以后，该队列是否自动删除，true自动删除，false不自动删除
         * param5 其他参数，暂时不用
         */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //从控制台输入消息
        Scanner sc = new Scanner(System.in);
        while (sc.hasNext()) {
            String message = sc.next();

            /**
             * 发送消息
             * param1 发送到哪个交换机
             * param2 路由的key值是哪个 ，本次是队列名称
             * param3 其他参数信息
             * param4 发送的消息体
             */
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("message send end : "+message);
        }


    }
}

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

消费者

两个消费者
worker03

package com.rabbitmqDemo.rabbitmq.three;

import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmqDemo.rabbitmq.utils.RabbitMqUtils;
import com.rabbitmqDemo.rabbitmq.utils.SleepUtils;

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

/**
 * 一个工作线程（相当于消费者）
 */
public class worker03 {
    //队列名称
    private static final String QUEUE_NAME = "ack_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMqUtils.getChannel();

        //声明 消费者成功消费的回调
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("worker03-message:" + new String(message.getBody()));
            System.out.println("worker03 sleep 2 second ......");

            SleepUtils.sleep(2);
            /**
             * param1 消息标记 tag
             * param2 是否批量应答消息
             */
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
        };
        //声明 取消消息时的回调
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("worker03-消息消费被中断-" + consumerTag);

        };
        /**
         * 消费者消费消息
         * param1 队列名称
         * param2 消费成功之后是否自动应答，true 代表自动应答，false表示不自动应答
         * param3 消费者成功消费的回调
         * param4 消费者取消消费回调
         */
        System.out.println("worker03等待接收消息......");
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, cancelCallback);

    }

}

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

worker04

package com.rabbitmqDemo.rabbitmq.three;

import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmqDemo.rabbitmq.utils.RabbitMqUtils;
import com.rabbitmqDemo.rabbitmq.utils.SleepUtils;

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

/**
 * 一个工作线程（相当于消费者）
 */
public class worker04 {
    //队列名称
    private static final String QUEUE_NAME = "ack_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMqUtils.getChannel();

        //声明 消费者成功消费的回调
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("worker04-message:" + new String(message.getBody()));
            System.out.println("worker04 sleep 30 second ......");

            SleepUtils.sleep(30);
            /**
             * param1 消息标记 tag
             * param2 是否批量应答消息
             */
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
        };
        //声明 取消消息时的回调
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("worker04-消息消费被中断-" + consumerTag);
        };
        /**
         * 消费者消费消息
         * param1 队列名称
         * param2 消费成功之后是否自动应答，true 代表自动应答，false表示不自动应答
         * param3 消费者成功消费的回调
         * param4 消费者取消消费回调
         */
        System.out.println("worker04等待接收消息......");
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, cancelCallback);

    }

}

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

SleeepUtils

package com.rabbitmqDemo.rabbitmq.utils;
/**
 * 休眠工具类
 */
public class SleepUtils {
    public static void sleep(int second) {
        try {
            Thread.sleep(second * 1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

执行过程

1. 生产者快速生产消息
   

2. 通过轮询 worker03 消费了消息，worker03休眠较短，消费了全部消息，并应答了
   

3. 通过轮询worker04消费了第一个消息，进入了休眠。并未及时应答
   

4. 在worker休眠期，结束worker04进行，消息没有进行应答
   

5. RabbitMQ会将本来分发给worker04的消息重新分发给worker03
   

<4>消息自动重新入队

1. 概念

如果消费者由于某些原因失去连接(其通道已关闭，连接已关闭或TCP连接丢失)，导致消息未发送ACK确认，RabbitMQ将了解到消息未完全处理，并将对其重新排队。如果此时其他消费者可以处理，它将很快将其重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。

2. 示意图