RabbitMQ消费者限流实现消息处理优化

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
        <version>2.5.0</version>
    </dependency>
</dependencies>

/**
 * 消费者的限流机制
 *  1、确保Ack机制为手动机制：acknowledge-mode: manual
 *  2、每次消费消息的个数：prefetch: 1 只有手动确认完后才会拉取下一条消息
 */
@Component
public class QosListener implements ChannelAwareMessageListener {
    @RabbitListener(queues = "test_queue_name")
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        System.out.println("消费者接受的消息为：" + new String(message.getBody()));
    }
}

/**
 * 消费者的限流机制
 *  1、确保Ack机制为手动机制：acknowledge-mode: manual
 *  2、每次消费消息的个数：prefetch: 1 只有手动确认完后才会拉取下一条消息
 */
@Component
public class QosListener implements ChannelAwareMessageListener {
    @RabbitListener(queues = "test_queue_name")
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        Thread.sleep(5000);
        long deliveryTag = message.getMessageProperties().getDeliveryTag();// 消息的唯一标识id
        System.out.println("消费者接受的消息为：" + new String(message.getBody()));
        channel.basicAck(deliveryTag,true);//每5s读一次消息（限流后每次为一条）
    }
}