我正在創(chuàng)建一個應(yīng)用程序，該應(yīng)用程序使用 RabbitMQ 向消費者發(fā)送消息以進行耗時的處理。但是，我需要對消息進行優(yōu)先級排序。當高優(yōu)先級的消息到達時，即使所有消費者實例都在處理其他消息，也必須處理該消息。AFAIK 在 Spring Boot 和 RabbitMQ 中不可能搶占處理低優(yōu)先級消息并切換到處理高優(yōu)先級消息。是否可以創(chuàng)建僅接受高優(yōu)先級消息的消費者，或者在所有其他繁忙且高優(yōu)先級消息到達時動態(tài)運行額外的消費者組？我嘗試使用 x-max-priority=10 標志添加隊列并增加消費者數(shù)量，但這并不能解決我的問題。想象一下，我們運行 50 個消費者并發(fā)送 50 條低優(yōu)先級消息。當執(zhí)行耗時的處理時，一條新消息以高優(yōu)先級到達，但無法立即處理，因為所有 50 個消費者都忙。配置中有一部分設(shè)置了消費者的數(shù)量@Beanpublic SimpleRabbitListenerContainerFactoryrabbitListenerContainerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, @Qualifier("rabbitConnectionFactory") ConnectionFactory connectionFactory) { SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); configurer.configure(factory, connectionFactory); factory.setConcurrentConsumers(50); factory.setMaxConcurrentConsumers(100); return factory;}有沒有辦法創(chuàng)建一組接受消息高優(yōu)先級消息（例如高于0）的消費者或為高優(yōu)先級消息動態(tài)創(chuàng)建消費者？
2 回答
手掌心
TA貢獻1942條經(jīng)驗 獲得超3個贊
我不知道實施您所描述的先發(fā)制人策略的方法，但是您可以考慮許多其他選擇。
優(yōu)先級設(shè)定
首先要考慮的是priorityRabbitMQ 本身的支持。
考慮一下 Gavin M. Roy 的《RabbitMQ in Depth》中的摘錄：
“從 RabbitMQ 3.5.0 開始，優(yōu)先級字段已按照 AMQP 規(guī)范實現(xiàn)。它被定義為一個整數(shù)，可能的值為 0 到 9，用于隊列中的消息優(yōu)先級排序。根據(jù)規(guī)定，如果發(fā)布了優(yōu)先級為 9 的消息，隨后又發(fā)布了優(yōu)先級為 0 的消息，則新連接的消費者將先于優(yōu)先級為 9 的消息接收到優(yōu)先級為 0 的消息。
例如
rabbitTemplate.convertAndSend("Hello World!", message -> {
MessageProperties properties = MessagePropertiesBuilder.newInstance()
.setPriority(0)
.build();
return MessageBuilder.fromMessage(message)
.andProperties(properties)
.build();
});
基于優(yōu)先級的交換
第二種選擇是定義主題交換并定義考慮您的優(yōu)先級的路由鍵。
例如，考慮events使用模式的路由密鑰進行交換，EventName.Priority例如OrderPlaced.High、OrderPlaced.Normal或OrderPlaced.Low。
基于此，您可以有一個隊列僅綁定到高優(yōu)先級的訂單，即OrderPlaced.High和多個專用于該隊列的消費者。
例如
String routingKey = String.format("%s.%s", event.name(), event.priority());
rabbit.convertAndSend(routingKey, event);
使用如下所示的偵聽器，其中隊列使用路由鍵high-priority-orders綁定到events事件OrderPlaced和優(yōu)先級的交換器。HighOrderPlaced.High
@Component
@RabbitListener(queues = "high-priority-orders", containerFactory="orders")
public class HighPriorityOrdersListener {
@RabbitHandler
public void onOrderPlaced(OrderPlacedEvent orderPlaced) {
//...
}
}
顯然，您將需要一個專用的線程池（在orders上面的容器工廠中）來處理高優(yōu)先級請求。
慕少森
TA貢獻2019條經(jīng)驗 獲得超9個贊
AMQP 協(xié)議中沒有從隊列中“選擇”消息的機制。
您可能需要考慮使用具有專用消費者的離散隊列。
順便說一句，這與 spring 無關(guān)；有關(guān) RabbitMQ 的一般問題應(yīng)直接咨詢rabbitmq-users Google 群組。
添加回答
舉報
0/150
提交
取消