RabbitMQ--消息应答

た入场券 2024-03-30 14:10 67阅读 0赞

## 一、概念 ##

#### 1、概念 ####

消费者完成一个任务可能需要一段时间，如果其中一个消费者处理一个长的任务并只完成了部分突然他就挂了，此时消息就会丢失。为了保证消息在发送过程中不丢失，RabbitMQ引入了消息应答机制，消息应答就是：消费者在接收到消息并处理该消息后，告诉RabbitMQ已经处理，RabbitMQ就可以把该消息删除。

#### 2、自动应答 ####

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

#### 3、手动应答 ####

*  Channel.basicAck()

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

*  Channel.basicNack()

用于否定回答

*  Channel.basicReject()

用于否定回答，不处理该消息直接拒绝，可以将其丢弃

解释：Channel.basicAck(参数1，参数2)

参数2：是否批量应答。

true：批量应答channel上的所有未应答的消息。

false：只应答当前消息。

#### 4、自动重新入队 ####

如果消费者由于某些原因失去连接，导致消息未发送ACK确认，RabbitMQ将对其重新排队。如果此时其他消费者可以处理，它将重新发送给另一个消费者。

## 二、手动应答代码 ##

目的：消息在手动应答时不丢失。

操作：启动两个消费者，在生产者生产消息的时候关闭其中一个消费者。

#### 1、消费者 ####

//消费者1号
    public class workerOne {
        //队列名称
        public static final String QUEUE_NAME="hello";
    
        public static void main(String[] args) {
            //创建连接
            Connection connection = ConnectionUtil.getConnection();
    
            try {
                Channel channel = connection.createChannel();
                //消费者未成功消费的回调
                DeliverCallback deliverCallback=(var1, var2)->{
                    System.out.println("接收到的消息："+new String(var2.getBody()));
    
                    try {
                        Thread.sleep(1000*10);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    //手动回应
                    channel.basicAck(var2.getEnvelope().getDeliveryTag(),false);
                    System.out.println("处理完成");
                };
    
                //取消消费
                CancelCallback cancelCallback= var->{
                    System.out.println("取消消费消息");
                };
                System.out.println("消费者2准备接收消息");
                //开启手动应答，参数2变为false
                channel.basicConsume(QUEUE_NAME,false,deliverCallback,cancelCallback);
    
            } catch (IOException e) {
                e.printStackTrace();
            }
    
        }
    
    }

#### 2、生产者 ####

public class WorkProducer {
    
        public static void main(String[] args) {
            //创建链接工厂、创建链接
            Connection connection= ConnectionUtil.getConnection();
            try {
                //获取信道
                Channel channel = connection.createChannel();
                /**
                 * 生产队列
                 * 1、队列名称
                 * 2、队列消息是否持久化
                 * 3、队列消息是否资源共享
                 * 4、是否自动删除
                 * 5、其他参数
                 * */
                channel.queueDeclare(workerOne.QUEUE_NAME,false,false,false,null);
                //发送的信息
                String msg="111";
                /**
                 * 发送消息
                 * 1、交换机名称
                 * 2、路由键
                 * 3、其他参数
                 * 4、消息体
                 * */
                    channel.basicPublish("",workerOne.QUEUE_NAME,null,msg.getBytes());
                    System.out.println("发送成功");
            } catch (IOException e) {
                System.out.println("发送失败");
                e.printStackTrace();
            }
        }
    }