RabbitMq高級特性之消費端限流 通俗易懂 超詳細 【內含案例】

RabbitMq高級特性之消費端限流

介紹

消息隊列中囤積了大量的消息, 或者某些時刻生產的消息遠遠大於消費者處理能力的時候, 這個時候如果消費者一次取出大量的消息, 但是客戶端又無法處理, 就會出現問題, 甚至可能導致服務崩潰, 所以需要對消費端進行限流

代碼展示

一丶首先部署SpringBoot框架

1. 完成 SpringBoot 整合 RabbitMq 中的Topic通配符模式

二丶在 resource資源文件夾里application.yml文件中 添加配置

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual #開啟手動簽收
        prefetch: 3 #一次就收三條

三、更改ProducerTest.java文件

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import javax.annotation.Resource;

@RunWith(SpringRunner.class)
@SpringBootTest
@Slf4j
public class producerTest {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Test
    public void test(){
    
        String routingKey = "item.insert";
    
        int count = 1;
        while (count <= 9){
            String message = "發送第"+count+"條消息";
            //log.debug("路由鍵:{}",routingKey);
            rabbitTemplate.convertAndSend(RabbitConfig.TOPIC_EXCHANGE_NAME,routingKey,message);
            count++;
        }
        log.debug("發送成功");
    }
}

四、更改CounmerListener.java文件

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;

/**
 * 消費者 消費監聽器
 */
@Component
@RabbitListener(queues = "direct_queue")
@Slf4j
public class ConsumerListener {

    @RabbitHandler
    public void accept(@Payload String message, @Headers Map map, Channel channel){

        long deliveryTag = (long) map.get(AmqpHeaders.DELIVERY_TAG);
        log.debug("deliveryTag:{}",deliveryTag);
        log.debug("message:{}",message);
        if (deliveryTag % 3 == 0) {
            try {
                //確認收到
                channel.basicAck(deliveryTag,true);
                Thread.sleep(3000);
                log.debug("休息三秒然后在接受消息");
            } catch (InterruptedException | IOException e) {
                e.printStackTrace();
            }
        }
    }

}

五、測試

1. 先運行測試文件 創建交換機和隊列
2. 然后在運行消息監聽器

六、聲明

本次內容運用到 RabbitMq確認消息機制