Как я могу контролировать скорость, с которой Spring получает из очереди?

Question

Я использую управляемую сообщениями среду POJO от Spring (и DefaultMessageListenerContainer в частности) для прослушивания нескольких очередей и тем.

В случае одной конкретной очереди необходимо замедлить скоростьв котором я истощаю очередь, порядка одного сообщения каждые пять минут.Фактическая обработка сообщений - это операция за секунду, но я бы хотел, чтобы слушатель некоторое время бездействовал между сообщениями.

Я создал небольшой взлом, но это явно не так.Оптимальный: то, что я сделал, это установил максимальный параллелизм равным 1 и добавил Thread.sleep(..) после обработки каждого сообщения.Вместо этого я хотел бы найти способ использовать DefaultMessageListenerContainer для ожидания между попытками получения, а не заставлять обработчик ожидать во время возможной обработки сообщения.

Я подумал, если бы был ScheduledExecutor, который бы помог, но я понимаю, что регулирование нужно будет выполнять там, где выполняются задачи.Может быть, есть какой-то метод из DefaultMessageListenerContainer, который я мог бы переопределить, чтобы выполнить то, что мне нужно?

Answer 1

В зависимости от поставщика очереди, вы можете установить максимальную скорость для потребителей, потребляющих ее очереди.

Например, в hornetQ вы устанавливаете это на фабрике соединений с помощью customer-max-rate.

Answer 2

Альтернативой изменению поведения вашего потребителя может быть использование Apache Camel для задержки сообщений в этой конкретной очереди.

http://camel.apache.org/delayer.html описывает функциональные возможности шаблона Camel Delayer. Так, например:

<route>
    <from uri="jms:YOURQUEUE"/>
    <delay>
        <constant>1000</constant>
    </delay>
    <to uri="jms:DELAYEDQUEUE"/>
</route>

Где вы бы тогда использовали DELAYEDQUEUE, и все сообщения будут задерживаться на 1 секунду.

Answer 3

Я не уверен на 100%, но верю, что receiveTimeout - это то, что вам нужно.

 <bean id="blahContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    ....
    <!-- 300000 = 5 * 60 * 1000 == 5 min -->
    <property name="receiveTimeout" value="300000"/>
</bean>

receiveTimeout принимает тайм-аут в миллисекундах, вы можете прочитать об этом в Javadocs