RabbitMQ消费方式汇总

Posted 2021-01-25 duanjt

在学习本章节前，请先学习之前的章节：
Java访问RabbitMQ：https://www.cnblogs.com/duanjt/p/10057330.html
RabbitMQ消息发布时的权衡：https://www.cnblogs.com/duanjt/p/10075308.html

一、推送Consume

前面我们使用到的都是这种模式，注册一个消费者后，RabbitMQ会在消息可用时，自动将消息进行推送给消费者。这种方式效率最高最及时。
核心代码如下：

// 接收消息,第二个参数表示是否自动应答
channel.basicConsume(queueName, true, new DefaultConsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
        System.out.println(envelope.getRoutingKey() + " 接收到数据:" + new String(body));
    }
});

 

 

二、拉取Get

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

while(true){
    //如果没有消息，将返回null
    GetResponse getResponse = channel.basicGet(queueName, true);    
    if(null!=getResponse){
        System.out.println("received["+getResponse.getEnvelope().getRoutingKey()+"]"+new String(getResponse.getBody()));
    }
    Thread.sleep(1000);
}

 

 

三、自动确认

方法channel.basicConsume和方法channel.basicGet表示同步或异步获取消息，第二个参数都表示是否自动确认。前面我们都设置为了true。这个时候我们只需要处理逻辑，将自动向RabbitMQ进行确认。
当autoAck=true时，一旦消费者接收到了消息，就视为自动确认了消息。如果消费者在处理消息的过程中，出了错，就没有什么办法重新处理这条消息，所以我们很多时候，需要在消息处理成功后，再确认消息，这就需要手动确认。

 

四、手动确认

 

// 接收消息手动确认,第二个参数表示是否自动应答
channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
        System.out.println(envelope.getRoutingKey() + " 接收到数据:" + new String(body));
        //手动确认，第一个参数是消息标识，第二个参数表示是否批量确认。这里是一条一条确认，所以设置false
        channel.basicAck(envelope.getDeliveryTag(), false);
    }
});

 

 

五、QoS预取模式

在确认消息被接收之前，消费者可以预先要求接收一定数量的消息，在处理完一定数量的消息后，批量进行确认。如果消费者应用程序在确认消息之前崩溃，则所有未确认的消息将被重新发送给其他消费者。所以这里存在着一定程度上的可靠性风险。
这种机制一方面可以实现限速（将消息暂存到RabbitMQ内存中）的作用，一方面可以保证消息确认质量（比如确认了但是处理有异常的情况）
核心代码：

//参数1表示限制条数，参数2 true=channel，false=消费者
channel.basicQos(100, true);

// 接收消息手动确认,第二个参数表示是否自动应答
channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
        System.out.println(envelope.getRoutingKey() + " 接收到数据:" + new String(body));
        //手动确认，第一个参数是消息标识，第二个参数表示是否批量确认。这里是一条一条确认，所以设置false
        channel.basicAck(envelope.getDeliveryTag(), false);
    }
});

 

 

注意：
1.消费确认模式必须是非自动ACK机制（这个是使用baseQos的前提条件，否则会Qos不生效），然后设置basicQos的值；另外，还可以基于consume和channel的粒度进行设置（global）