rabbitMQ学习-消息应答

消息应答

为了防止消息在发送过程中不丢失，他是指消费者收到消息并且处理该消息之后，告诉rabbitmq他已经处理完成了，rabbitmq可以把消息删除。
自动应答：

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

手动应答

能用手动则用手动。手动方法

1. Channel.basicAck(用于肯定确认)
2. Channel.basicNack(用于否定确认)
3. Channel.basicReject(用于否定确认)，对比nack少了一个参数，不处理该消息了直接就可以拒绝，也可以将他丢弃。

批量应答（multiple）：推荐使用false,当接受到一个消息的时候，也就是队列往信道里边放了n条数据，那么他会将此信道里边的所有消息都应答一次。

消息自动重新入队

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

个人理解：

也就是本来做这个任务的员工有事辞职了，公司为了完成任务交给另一个还在职的员工完成这个任务，直到这个任务完成，这样保证任务不被丢失。

代码实现：

消息在手动应答的时候是不应该被丢失的，且会重新入队。

/*
我们主要围绕的是手动应答进行处理，要是消息应答不丢失，放回队列
重新消费
 */
public class Task {
    //老样子，队列名称
    public static  final  String QUEUE_NAME = "ack_queue";

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

        //声名队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);

        //从控制台中获取
        Scanner sc = new Scanner(System.in);
        while (sc.hasNext()){
            String message = sc.next();
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes("UTF-8"));
            System.out.println("生产者发送消息： "+ 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

/*
接收消息
 */
public class Work_Consume1 {

    public static  final  String QUEUE_NAME = "ack_queue";


    public static void main(String[] args) throws IOException, TimeoutException {
        //接受消息
        Channel channel = GetConnection.getChannel();

        System.out.println("C1等待接受消息处理时间较短");

        DeliverCallback deliverCallback = (consumerTag, message)->{
            //看情况你给不给休眠
            SleepUtils.sleep(1);

            System.out.println("接收到消息" + new String(message.getBody(),"UTF-8"));
            //进行手动应答
            /*
            参数1 ， 消息的标记tag
            参数2， 是否批量应答  false 不 ，true，批量应答
             */
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
        };
        //手动接收消息


        //采用手动应答
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME,autoAck,deliverCallback,(consumerTag->{
            System.out.println(consumerTag + "消费者取消消费接口回调逻辑");
        }));



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

/*
 * @Title 休眠工具类
 */
public class SleepUtils {
    public static  void sleep(int second) {
        //JUC里边的工具类

        try {
            Thread.sleep(1000*second);
            //TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            //中断
            Thread.currentThread().interrupt();
        }
    }

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

通过上述方式，我们可以知道消息队列在我们创建多个消费者的时候，是轮转来的，当然，假设有两个消费者，如果有一个宕机，那么消息队列就将本轮转给没宕机的那个消费者消费，只要生产者有产生，无论如何要将消息全部消费，不可丢弃。