rabbitmq消费者---------消息的获取和ack确认

た入场券 2022-12-22 06:18 249阅读 0赞

## 消息的获取方式 ##

### 1、拉取get---------------不建议使用 ###

属于一种轮询模型，发送一次 get 请求，获得一个消息。如果此时 RabbitMQ 中没有消息，会获得一个表示空的回复。总的来说，这种方式性能比较  
差，很明显，每获得一条消息，都要和 RabbitMQ 进行网络通信发出请求。而且对 RabbitMQ 来说，RabbitMQ 无法进行任何优化，因为它永远不知道应用  
程序何时会发出请求。

while(true){
                //拉一条，自动确认的(rabbit 认为这条消息消费 -- 从队列中删除)
                GetResponse getResponse = channel.basicGet(queueName, false);
                if(null!=getResponse){
                    System.out.println("received["
                            +getResponse.getEnvelope().getRoutingKey()+"]"
                            +new String(getResponse.getBody()));
                }
                //确认(自动、手动)
                channel.basicAck(0,true);
                Thread.sleep(1000);
            }

### 2、消息的推送consume-------------最常用的模式 ###

属于一种推送模型。注册一个消费者后，RabbitMQ 会在消息可用时，自动将消息进行推送给消费者。

//申明一个消费者
            final Consumer consumer  = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String s, Envelope envelope, AMQP.BasicProperties     basicProperties, byte[] bytes) throws IOException {
                    String message = new String(bytes,"UTF-8");
                    System.out.println("Received["+envelope.getRoutingKey()+"]"+message);
                }
            };
            //消息者正是开始在指定队列上消费。(queue-king)
            //TODO 这里第二个参数是自动确认参数，如果是true则是自动确认
            channel.basicConsume(queueName,true,consumer);

## 消息的应答 ##

消费者收到的每一条消息都必须进行确认。消息确认后，RabbitMQ 才会从队列删除这条消息，RabbitMQ 不会为未确认的消息设置超时时  
间，它判断此消息是否需要重新投递给消费者的唯一依据是消费该消息的消费者连接是否已经断开。这么设计的原因是 RabbitMQ 允许消费者消费一条消  
息的时间可以很久很久。

### 自动确认 ###

消费者在声明队列时，可以指定 autoAck 参数，当 autoAck=true 时，一旦消费者接收到了消息，就视为自动确认了消息。如果消费者在处理消息的过  
程中，出了错，就没有什么办法重新处理这条消息，所以我们很多时候，需要在消息处理成功后，再确认消息，这就需要手动确认。

//TODO 这里第二个参数是：是否需要自动确认参数，如果是false则是手动确认，如果是true则是自动确认
            channel.basicConsume(queueName,true,consumer);

### 手动确认 ###

当 autoAck=false 时，RabbitMQ 会等待消费者显式发回 ack 信号后才从内存(和磁盘，如果是持久化消息的话)中移去消息。否则，RabbitMQ 会在队列  
中消息被消费后立即删除它。  
采用消息确认机制后，只要令 autoAck=false，消费者就有足够的时间处理消息(任务)，不用担心处理消息过程中消费者进程挂掉后消息丢失的问题，  
因为 RabbitMQ 会一直持有消息直到消费者显式调用 basicAck 为止。  
当 autoAck=false 时，对于 RabbitMQ 服务器端而言，队列中的消息分成了两部分：一部分是等待投递给消费者的消息；一部分是已经投递给消费者，  
但是还没有收到消费者 ack 信号的消息。如果服务器端一直没有收到消费者的 ack 信号，并且消费此消息的消费者已经断开连接，则服务器端会安排该消  
息重新进入队列，等待投递给下一个消费者（也可能还是原来的那个消费者）。

/*声明了一个消费者*/
            final Consumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag,
                                           Envelope envelope,
                                           AMQP.BasicProperties properties,
                                           byte[] body) throws IOException {
                    String message = new String(body, "UTF-8");
                    System.out.println("Received["+envelope.getRoutingKey()
                            +"]"+message);
                    //TODO 这里进行确认
                    System.out.println("手动确认的tag:"+envelope.getDeliveryTag());
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            };
            /*消费者正式开始在指定队列上消费消息*/
            //TODO 这里第二个参数是自动确认参数，如果是false则是手动确认
            channel.basicConsume(queueName,false,consumer);

通过运行程序，启动两个消费者 A、B，都可以收到消息，但是其中有一个消费者 A 不会对消息进行确认，当把这个消费者 A 关闭后，消费者 B 又会  
收到本来发送给消费者 A 的消息。所以我们一般使用手动确认的方法是，将消息的处理放在 try/catch 语句块中，成功处理了，就给 RabbitMQ 一个确认应  
答，如果处理异常了，就在 catch 中，进行消息的拒绝，如何拒绝，参考消息的拒绝章节[https://blog.csdn.net/nandao158/article/details/109801226][https_blog.csdn.net_nandao158_article_details_109801226]。

### QoS 预取模式 ###

在确认消息被接收之前，消费者可以预先要求接收一定数量的消息，在处理完一定数量的消息后，批量进行确认。如果消费者应用程序在确认消息  
之前崩溃，则所有未确认的消息将被重新发送给其他消费者。所以这里存在着一定程度上的可靠性风险。  
这种机制一方面可以实现限速（将消息暂存到 RabbitMQ 内存中）的作用，一方面可以保证消息确认质量（比如确认了但是处理有异常的情况）。  
注意：消费确认模式必须是非自动 ACK 机制（这个是使用 baseQos 的前提条件，否则会 Qos 不生效），然后设置 basicQos 的值；另外，还可以基于  
consume 和 channel 的粒度进行设置（global）。

生产者代码（主要是模拟生产者发送较多的代码）

//TODO 生产者发送非常多的数据
            //发送210条消息，其中第210条消息表示本批次消息的结束
            for(int i=0;i<210;i++){
                // 发送的消息
                String message = "Hello World_"+(i+1);
                if(i==209){ //最后一条
                    message = "stop";
                }
                //参数1：exchange name
                //参数2：routing key
                channel.basicPublish(EXCHANGE_NAME, "error",
                        null, message.getBytes());
                System.out.println(" [x] Sent 'error':'"
                        + message + "'");
            }

QOS预取代码

//TODO 150条预取(150都取出来 150， 210-150  60  )
            channel.basicQos(150,true);
            /*消费者正式开始在指定队列上消费消息*/
            channel.basicConsume(queueName,false,consumer);

QOS预取代码-----------自定义批量确认

1、实现DefaultConsumer

package cn.enjoyedu.consumer_balance.qos;
    
    import com.rabbitmq.client.AMQP;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    
    import java.io.IOException;
    
    /**
     * 批量确认 -----消费者
     */
    public class BatchAckConsumer extends DefaultConsumer {
        //计数，第多少条
        private  int meesageCount =0;
        public BatchAckConsumer(Channel channel) {
            super(channel);
            System.out.println("批量消费者启动了......");
        }
    
        @Override
        public void handleDelivery(String consumerTag,
                                   Envelope envelope,
                                   AMQP.BasicProperties properties,
                                   byte[] body) throws IOException {
            //把消息体拉出来
            String message = new String(body,"UTF-8");
    
            System.out.println("批量消费者---Received["+envelope.getRoutingKey()
                    +"]"+message);
            meesageCount++;
            //批量确认 50一批
            if(meesageCount %50 ==0){
               this.getChannel().basicAck(envelope.getDeliveryTag(),true);
                System.out.println("批量消息费进行消息的确认------------");
            }
            if(message.equals("stop")){ //如果是最后一条消息，则把剩余的消息都进行确认
                this.getChannel().basicAck(envelope.getDeliveryTag(),true);
                System.out.println("批量消费者进行最后业务消息的确认---------");
            }
        }
    }

2、消费者的代码

//TODO 150条预取(150都取出来 150， 210-150  60  )
            channel.basicQos(150,true);
            /*消费者正式开始在指定队列上消费消息*/
            //TODO 自定义消费者批量确认
            BatchAckConsumer batchAckConsumer = new BatchAckConsumer(channel);
            channel.basicConsume(queueName,false,batchAckConsumer);

[https_blog.csdn.net_nandao158_article_details_109801226]: https://blog.csdn.net/nandao158/article/details/109801226