是否可以选择创建只接受RabbitMQ和Spring中优先级较高的消息的使用者？

Question

我正在创建一个应用程序，该应用程序使用RabbitMQ向消费者发送用于时间昂贵的处理的消息。然而，我需要优先考虑信息。当具有高优先级的消息到达时，即使所有使用者实例都在处理其他消息，也必须对其进行处理。

在Spring和RabbitMQ中，不可能抢先处理低优先级消息并切换到处理高优先级消息。

是否有可能创建只接受高优先级消息的使用者，或者在所有其他繁忙且高优先级的消息到达时动态运行附加的一组使用者？

我试图添加带有x优先级=10标志的队列，并增加用户数量，但这并没有解决我的问题。

假设我们运行50个消费者并发送50个优先级较低的消息。当执行时间昂贵的处理时，一条新消息以高优先级到达，但由于50名消费者都很忙，所以不能立即处理。

配置中有一部分设置了使用者的数量。

@Bean
public SimpleRabbitListenerContainerFactory
rabbitListenerContainerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer,
                               @Qualifier("rabbitConnectionFactory") ConnectionFactory connectionFactory) {
  SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
  configurer.configure(factory, connectionFactory);
  factory.setConcurrentConsumers(50);
  factory.setMaxConcurrentConsumers(100);
  return factory;
}

是否有一种方法可以创建一组接受高优先级消息(例如，高于0)的使用者，或者为高优先级的消息动态创建使用者？

Answer 1

我不知道如何实现你所描述的先发制人的策略，但是你可以考虑一些其他的事情。

优先级设置

首先要考虑的是RabbitMQ本身对RabbitMQ的支持。

考虑一下加文·M·罗伊(GavinM.Roy)从RabbitMQ摘录的深度：

“在RabbitMQ 3.5.0中，优先级字段已经按照AMQP规范实现。它被定义为一个整数，其可能值为0到9，用于队列中的消息优先级。如前所述，如果发布优先级为9的消息，随后发布优先级为0的消息，则新连接的使用者将在优先级为9的消息之前接收优先级为0的消息“。

例如：

rabbitTemplate.convertAndSend("Hello World!", message -> {
  MessageProperties properties = MessagePropertiesBuilder.newInstance()
                                                         .setPriority(0)
                                                         .build();
  return MessageBuilder.fromMessage(message)
                       .andProperties(properties)
                       .build();
});

基于优先级的交换

第二种选择是定义主题交换，并定义考虑优先级的路由密钥。

例如，考虑使用模式events的路由密钥交换EventName.Priority，例如OrderPlaced.High、OrderPlaced.Normal或OrderPlaced.Low。

基于此，您可以有一个仅绑定到高优先级订单的队列，即OrderPlaced.High和许多专门用于该队列的用户。

例如：

String routingKey = String.format("%s.%s", event.name(), event.priority());
rabbit.convertAndSend(routingKey, event);

对于下面这样的侦听器，队列high-priority-orders使用路由密钥OrderPlaced.High绑定到事件OrderPlaced和优先级High的events交换。

@Component
@RabbitListener(queues = "high-priority-orders", containerFactory="orders")
public class HighPriorityOrdersListener {

 @RabbitHandler
 public void onOrderPlaced(OrderPlacedEvent orderPlaced) {
   //...
 }
}

显然，您需要一个专门的线程池(在上面的orders容器工厂中)来处理高优先级请求。

Answer 2

AMQP协议中没有从队列中“选择”消息的机制。

您可能需要考虑将离散队列与专用的使用者一起使用。

顺便说一句，这与春季无关；关于RabbitMQ的一般问题应该直接指向狂犬病用户Google组。