Блог

Мы используем RabbitMQ с конфигурациями весенней загрузки по умолчанию. У нас есть вариант использования, в котором мы не хотим параллелизма для одного из слушателей. То есть мы хотим, чтобы в любой данный момент времени работал только один поток потребителя. Мы хотим этого, потому что природа варианта использования такова, что мы хотим, чтобы сообщения обрабатывались по порядку, поэтому, если на одного потребителя приходится несколько потоков, может быть вероятность того, что сообщения обрабатываются не по порядку. Поскольку мы используем значения по умолчанию и явно не изменили контейнер, мы используем SimpleMessageListenerContainer . Просмотрев документацию, я попытался зафиксировать количество пользователей, которые ее используют concurrency = "1" . Аннотация целевого метода выглядит следующим @RabbitListener(queues = ["queue-name"], concurrency = "1") образом .

Согласно документации, это должно было гарантировать, что существует только поток потребителей.

{@ссылка org.springframework.amqp.кролик.слушатель.SimpleMessageListenerContainer * SimpleMessageListenerContainer} если это значение является простым целым числом, оно задает фиксированное * количество потребителей в свойстве {@code concurrentConsumers}

 2021-10-29 06:11:26.361 INFO  29752 --- [ntContainer#4-1] c.t.t.i.p.s.xxx         : Created xxx
2021-10-29 06:11:26.383 INFO  29752 --- [ntContainer#0-1] c.t.t.i.p.s.xxx         : Created xxx
 

Идентификаторы потоков, которые следует отметить здесь, являются [ntContainer#4-1] и [ntContainer#0-1] .

Итак, вопрос в том, как мы можем гарантировать, что на каждого потребителя в любой данный момент времени приходится только один поток ?

Изменить: Добавление кода потребительского класса для большего контекста

 @ConditionalOnProperty(value = ["rabbitmq.sharebooking.enabled"], havingValue = "true", matchIfMissing = false)
class ShareBookingConsumer @Autowired constructor(
    private val shareBookingRepository: ShareBookingRepository,
    private val objectMapper: ObjectMapper,
    private val shareDtoToShareBookingConverter: ShareBookingDtoToShareBookingConverter
) {
    private val logger = LoggerFactory.getLogger(javaClass)

    init {
        logger.info("start sharebooking created consumer")
    }

    @RabbitListener(queues = ["tax_engine.share_booking"], concurrency = "1-1", exclusive = true)
    @Timed
    @Transactional
    fun consumeShareBookingCreatedEvent(message: Message) {
        try {
            consumeShareBookingCreatedEvent(message.body)
        } catch (e: Exception) {
            throw AmqpRejectAndDontRequeueException(e)
        }
    }

    private fun consumeShareBookingCreatedEvent(event: ByteArray) {
        toShareBookingCreationMessageEvent(event).let { creationEvent ->
            RmqMetrics.measureEventMetrics(creationEvent)
            val shareBooking = shareDtoToShareBookingConverter.convert(creationEvent.data)
            val persisted = shareBookingRepository.save(shareBooking)
            logger.info("Created shareBooking ${creationEvent.data.id}")
        }
    }

    private fun toShareBookingCreationMessageEvent(event: ByteArray) =
        objectMapper.readValue(event, shareBookingCreateEventType)

    companion object {
        private val shareBookingCreateEventType =
            object : TypeReference<RMQMessageEnvelope<ShareBookingCreationDto>>() {}
    }
}

Edit: Adding application thread analysis using visualvm
5 threads get created for 5 listeners.

  [1]: https://i.stack.imgur.com/gQINE.png
 

Набор concurrency = "1-1" . Обратите внимание , что параллелизм Прослушивателя зависит не только от concurrentConsumers , но и от maxConcurrentConsumers :

Если вы используете настраиваемую фабрику:

   @Bean
  public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(CachingConnectionFactory cachingConnectionFactory) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(cachingConnectionFactory);
    factory.setConcurrentConsumers(1);
    factory.setMaxConcurrentConsumers(1);
    return factory;
  }

 

См.: https://docs.spring.io/spring-amqp/docs/current/reference/html/#simplemessagelistenercontainer для деталей.

РЕДАКТИРОВАТЬ: Я провел простой тест, 2 потребителя и 2 потока:

  @RabbitListener(queues = "myQueue111", concurrency = "1-1")
  public void handleMessage(Object message) throws InterruptedException {
    LOGGER.info("Received message : {} in {}", message, Thread.currentThread().getName());
  }

  @RabbitListener(queues = "myQueue222", concurrency = "1-1")
  public void handleMessag1e(Object message) throws InterruptedException {
    LOGGER.info("Received message222 : {} in {}", message, Thread.currentThread().getName());
  }
 

Попробуй это:

 @RabbitListener(queues = ["queue-name"], exclusive = true)
 

Видишь https://docs.spring.io/spring-amqp/docs/current/reference/html/#exclusive-consumer