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();
            }
        }
    }

}

 

五、測試

先執行測試檔案 建立交換機和佇列

然後在執行訊息監聽器

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