Отложенная обработка событий с помощью RabbitMQ Delayed Message

Хочу поделиться очень интересным и простым способом отложенной обработки событий. Для чего это нужно? Простой пример - вам нужно уведомить сторонний сервис о неком событии. Если он не подтвердил получение вашего уведомления, то надо уведомить его еще раз через X минут. Эту задачу очень просто решить с помощью RabbitMQ.

RabbitMQ по умолчанию не умеет откладывать сообщения, они доставляются сразу после публикации. Функционал отложенной доставки доступен в виде плагина rabbitmq-delayed-message-exchange.



Обратите внимание, что для использования плагина требуется Erlang/OTP версии 18.0 и выше.

Включается плагин с помощью команды:

$ rabbitmq-plugins enable rabbitmq_delayed_message_exchange

Для использования отложенно доставки сообщений нужно объявить exchange типом x-delayed-message:

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-delayed-type", "direct");
channel.exchangeDeclare("my-exchange", "x-delayed-message", true, false, args);

При публикации сообщений в заголовке x-delay передается время (мс) на которое нужно задержать сообщение:

byte[] messageBodyBytes = "delayed payload".getBytes("UTF-8");
Map<String, Object> headers = new HashMap<String, Object>();
headers.put("x-delay", 5000);
AMQP.BasicProperties.Builder props = new AMQP.BasicProperties.Builder().headers(headers);
channel.basicPublish("my-exchange", "", props.build(), messageBodyBytes);

byte[] messageBodyBytes2 = "more delayed payload".getBytes("UTF-8");
Map<String, Object> headers2 = new HashMap<String, Object>();
headers2.put("x-delay", 1000);
AMQP.BasicProperties.Builder props2 = new AMQP.BasicProperties.Builder().headers(headers2);
channel.basicPublish("my-exchange", "", props2.build(), messageBodyBytes2);

В примере выше через секунду после публикации будет доставлено сообщение "more delayed payload", а еще через четыре секунды сообщение "delayed payload".

Более подробнее про ограничения плагина читайте на его официальной странице. В следущей статье я покажу как использовать RabbitMQ Delayed Message Plugin в Spring Framework.