前言

目的主要是学习RabbitMQ的Qos限流，大概会简单介绍学习为主：毕竟还是要来演示Springboot整合RabbitMQ注解的方式来使用

一.消费端限流

假设一个场景，比如，由于之前Rabbitmq服务器积压了许多之前未被处理的上万条消息，当我随便打开其中一个消费者客户端，会出现这种问题，信息一涌而进，当数据量特别大的时候可能会导致服务器卡顿或者直接崩溃，于是我们应该对消费端限流，用于保持的稳定。

1.1 basicQos

RabbitMQ 提供了一种 qos （服务质量保证）功能，即在非自动确认消息的前提下，如果一定数目的消息（通过基于 consume 或者 channel 设置 Qos 的值）未被确认前，不进行消费新的消息。

自动签收要设置成false, 建议实际工作中也设置成false
void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException;
prefetchSize : 消息大小限制, 一般设置为0, 消费端不做限制
prefetchCount : 会告诉RabbitMQ不要同时给一个消费者推送多于N个消息, 即一旦有N个消息还没有ack, 则该consumer将block(阻塞), 直到有消息ack
global : true/false 是否将上面设置应用于channel, 简单来说就是上面的限制是channel级别的还是consumer级别
注意 :
prefetchSize和global这两项, RabbitMQ没有实现, 暂且不关注, prefetchCount在autoAck设置false的情况下生效, 即在自动确认的情况下这两个值是不生效的

1.2 对消费端进行限流

1)我们既然要使用消费端限流，我们需要关闭自动 ack，将 autoAck 设置为 false

channel.basicConsume(queueName, false, consumer);

2)我们来设置具体的限流大小以及数量。

// 0,15 从0到15限制15条
// 设置 false 应该于consumer级别 
channel.basicQos(0, 15, false);

在消费者的 handleDelivery 消费方法中手动 ack，并且设置批量处理 ack 回应为 true

channel.basicAck(envelope.getDeliveryTag(), true);

2 生产端和消费端工程配置

省略配置....可翻之前的Rabbitmq工程配置

3.生产端工程

config配置

3.1 config配置

声明队列，此处使用direct队列

@Component
@Slf4j
public class RabbitListenerConfig {

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);
        connectionFactory.setPassword("guest");
        connectionFactory.setUsername("guest");
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        connectionFactory.setPublisherReturns(true);
        connectionFactory.createConnection();
        return connectionFactory;
    }

    @Bean
    public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        return factory;
    }
    @Bean
    @Qualifier("rabbitTemplate")
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //开启mandatory模式（开启失败回调）
        rabbitTemplate.setMandatory(true);
        //添加失败回调方法
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.info("message:{}, replyCode:{}, replyText:{}, exchange:{}, routingKey:{}",
                    message, replyCode, replyText, exchange, routingKey);
        });
        // 添加发送方确认模式方法
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) ->
                log.info("correlationData:{}, ack:{}, cause:{}",
                        correlationData.getId(), ack, cause));
        return rabbitTemplate;
    }


    /***声明 direct 队列  一对一***/
    @Bean
    public Exchange directExchange(){
        return new DirectExchange("dircet.exchange.test");
    }

    @Bean
    public Queue directQueue(){
        return new Queue("direct.queue.test");
    }

    @Bean
    public Binding directBinding(){
        return new Binding("direct.queue.test",
                                Binding.DestinationType.QUEUE,
                                "dircet.exchange.test",
                                    "direct.key",null);
    }
}

3.2 dto

@Getter
@Setter
@ToString
public class OrderMessageDTO implements Serializable{
    private Integer orderId;
    private BigDecimal price;
    private Integer productId;
}

3.3 service

public interface DirectService {

public void sendMessage();

//  使用限流Qos
public void sendQosMessage() throws JsonProcessingException;

}

3.4 impl

此处使用了另一种传递消息，使用了MessageProperties，将MessageProperties传递的对象转换成message

@Slf4j
@Service
public class DirectServiceImpl implements DirectService {

    @Autowired
    private RabbitTemplate rabbitTemplate;


    ObjectMapper objectMapper = new ObjectMapper();

    /**
     * 发送Qos 限流消息
     */
    @Override
    public void sendQosMessage() throws JsonProcessingException {
        /* 使用MessageProperties传递的对象转换成message*/
        MessageProperties messageProperties = new MessageProperties();
        OrderMessageDTO orderMessageDTO = new OrderMessageDTO();
        orderMessageDTO.setProductId(100);
        orderMessageDTO.setPrice(new BigDecimal("20"));
        orderMessageDTO.setOrderId(1);
        String messageToSend = objectMapper.writeValueAsString(orderMessageDTO);
        Message message = new Message(messageToSend.getBytes(),messageProperties);
        // 发送端确认是否确认消费
        CorrelationData correlationData = new CorrelationData();
        // 唯一ID
        correlationData.setId(orderMessageDTO.getOrderId().toString());
        rabbitTemplate.convertAndSend("dircet.exchange.test","direct.queue.test",message,correlationData);
    }
}

3.5 controller

@RestController
@Slf4j
@RequestMapping("/api")
public class SendController {
    @Autowired
    private DirectService directService;

    @GetMapping("/qosDirect")
    public void qosDirect() throws JsonProcessingException {
        for (int i = 0; i < 200; i++) {
            directService.sendQosMessage();
        }
    }
}

4.消费端工程

4.1 config配置

@Component
@Slf4j
public class RabbitListenerConfig {

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);
        connectionFactory.setPassword("guest");
        connectionFactory.setUsername("guest");
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        connectionFactory.setPublisherReturns(true);
        connectionFactory.createConnection();
        return connectionFactory;
    }

    @Bean
    public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        return factory;
    }
    
}

4.2 service

public interface DirectReciveService {
    // 处理监听生产端Qos限流
    public void QosDirectRecive(Message message, Channel channel) throws IOException;
}

4.3 impl

    @RabbitListener(
            containerFactory = "rabbitListenerContainerFactory",
            bindings = {
                    @QueueBinding(
                            value = @Queue(name = "direct.queue.test"),
                            exchange = @Exchange(name = "dircet.exchange.test",
                                    type = ExchangeTypes.DIRECT),
                            key = "direct.queue.test"
                    )
            }
    )
    @Override
    public void QosDirectRecive(@Payload Message message, Channel channel) throws IOException {
        log.info("========directQos接受消息===========");
        channel.basicQos(0,10,false);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                String message = new String(body, "UTF-8");
                log.info("[x] Received '" + message + "'");

                channel.basicAck(envelope.getDeliveryTag(), true);
            }
        };
        //. 设置 Channel 消费者绑定队列
        channel.basicConsume("direct.queue.test",false,consumer);
    }

5.演示

5.1 访问 http://localhost:8082/api/qosDirect

5.2 Rabbitmq控制台

image.png

上图中发现 Unacked值一直存在，每过 5 秒消费一条消息即 Ready 和 Total 都减少 10，而 Unacked的值在这里代表消费者正在处理的消息，通过我们的实验发现了消费者一次性最多处理 10 条消息，达到了消费者限流的预期功能。

Springboot与RabbitMQ上手学习之Qos限流(四)