RabbitMq高級特性之消費端限流
介紹
訊息佇列中囤積了大量的訊息, 或者某些時刻生產的訊息遠遠大于消費者處理能力的時候, 這個時候如果消費者一次取出大量的訊息, 但是客戶端又無法處理, 就會出現問題, 甚至可能導致服務崩潰, 所以需要對消費端進行限流
代碼展示
一丶首先部署SpringBoot框架
- 完成 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確認訊息機制
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/134843.html
標籤：Java