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