springboot整合rabbitmq 消费者Consumer 手动进行ack确认_rabbitmq中的basicconsume方法的使用,手动确定ack

ack指Acknowledge，确认。 表示消费端收到消息后的确认方式。
有三种确认方式：
自动确认：acknowledge="none"
手动确认：acknowledge="manual"
根据异常情况确认：acknowledge="auto"，（这种方式使用麻烦，不作讲解）

其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息缓存中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。

代码演示：用队列中的消息做除法运算，如果是整数，就会手动提交ack，如果是随便输入的文字，就会触发 basicNack

1 修改配置文件

server:
  port: 8021
spring:
  #给项目来个名字
  application:
    name: rabbitmq-test
  #配置rabbitMq 服务器
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: need
    password: 123456
    #虚拟host 可以不设置,使用server默认host
    virtual-host: /testhost
 
    #ack 确认方式
    listener:
      simple:
        acknowledge-mode: manual
      direct:
        acknowledge-mode: manual

相关的api解释

channel basicAck(long deliveryTag, boolean multiple);
deliveryTag:该消息的index
multiple：是否批量.true:将一次性ack所有小于deliveryTag的消息。

void basicNack(long deliveryTag, boolean multiple, boolean requeue);
deliveryTag:该消息的index
multiple：是否批量.true:将一次性拒绝所有小于deliveryTag的消息。
requeue：被拒绝的是否重新入队列

channel.basicReject(long deliveryTag, boolean requeue);
deliveryTag:该消息的index
requeue：被拒绝的是否重新入队列

channel.basicNack 与 channel.basicReject 的区别在于basicNack可以拒绝多条消息，而basicReject一次只能拒绝一条消息

2 消费者代码示例，有些是使用实现接口的方式 ，那是原生api的用法

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
 
@Component
public class DirectReceiver_1  {
 
    @RabbitListener(queues = "TestDirectQueue")//监听的队列名称 TestDirectQueue
    public void process(Message message, Channel channel) throws IOException {
 
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        //或 
        //@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag
 
        try {
            String msgbody = new String(message.getBody());
            //1.接收转换消息
            System.out.println("DirectReceiver消费者 1 收到消息  : " +msgbody+" 编号： "+deliveryTag);
 
            //2. 处理业务逻辑
            System.out.println("处理业务逻辑...");
            //模拟出现错误
            System.out.println(500/Double.valueOf(msgbody));
            //3. 手动签收
            channel.basicAck(deliveryTag,true);
        } catch (Exception e) {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException interruptedException) {
                interruptedException.printStackTrace();
            }
            //e.printStackTrace();
 
            //4.拒绝签收
            /*
            第三个参数：requeue：重回队列。如果设置为true，则消息重新回到queue，broker会重新发送该消息给消费端
             */
            channel.basicNack(deliveryTag,true,true);
            //channel.basicReject(deliveryTag,true);
        }
    }
}

3 随便输入的消息，如图，未确认ack