Повторяющиеся сообщения из очередей «Ответить» в другую очередь - PullRequest
Новая
       77

Повторяющиеся сообщения из очередей «Ответить» в другую очередь

0 голосов
/ 29 января 2020

Я использую rabbitMq в основном в режиме RP C, но я также хочу скопировать сообщения с запросами и ответами в другую очередь.

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

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

Но мне не удается найти способ "разветвить" сообщения, отправленные с помощью функции прямого ответа.

Пока я понимаю, что ответные сообщения отправляются по умолчанию прямой обмен с сгенерированным ключом routing_key в форме amqp.rabbitmq.reply-to.generatedName и, поскольку обмен по умолчанию неприкасаем, я не могу дублировать эти сообщения.

Знаете ли вы какой-либо способ сделать это ?

У меня есть решение, которого я бы предпочел избежать: заставить клиента повторно отправить ответ, полученный от reply-to, в «очередь журналирования».

но это означает, что мой клиент несет ответственность за эту функцию «ведения журнала», и я бы предпочел не делать этого.

кстати, даже если я не думаю, что это актуально, потому что это, вероятно, сервер rabbitMq проблема конфигурации, я использую Spring-AMQP клиент

Ответы [ 2 ]

1 голос
/ 30 января 2020

Вы не можете сделать это с фиксированным ответом, потому что нет реальной очереди / обмена.

Однако вы можете настроить каждый RabbitTemplate на использование фиксированной очереди ответов и контейнера ответов для прямые ответы из этой очереди в шаблон.

Документация здесь .

Кроме того, при использовании этого механизма вы можете настроить replyAddress шаблона в форме ключа обмена и маршрутизации. 

/**
 * An address for replies; if not provided, a temporary exclusive, auto-delete queue will
 * be used for each reply, unless RabbitMQ supports 'amq.rabbitmq.reply-to' - see
 * https://www.rabbitmq.com/direct-reply-to.html
 * <p>The address can be a simple queue name (in which case the reply will be routed via the default
 * exchange), or with the form {@code exchange/routingKey} to route the reply using an explicit
 * exchange and routing key.
 * @param replyAddress the replyAddress to set
 */
public synchronized void setReplyAddress(String replyAddress) {...}

Вы просто настраиваете шаблон и контейнер как обычно, делая шаблон слушателем контейнера ... 

@Bean
public RabbitTemplate amqpTemplate() {
    RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
    rabbitTemplate.setMessageConverter(msgConv());
    rabbitTemplate.setReplyAddress(replyQueue().getName());
    rabbitTemplate.setReplyTimeout(60000);
    rabbitTemplate.setUseDirectReplyToContainer(false);
    return rabbitTemplate;
}

@Bean
public SimpleMessageListenerContainer replyListenerContainer() {
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
    container.setConnectionFactory(connectionFactory());
    container.setQueues(replyQueue());
    container.setMessageListener(amqpTemplate());
    return container;
}

Опять же, это из документация.

0 голосов
/ 30 января 2020

Я нашел способ достичь того, чего хотел.

Это не так гибко, как предлагал Гари Рассел: { ссылка }

, но я могу использовать функцию пожарного шланга и связать очередь (или обмен на дополнительный контроль) на обмене amq.rabbitmq.trace с ключом маршрутизации, установленным в «publi sh». (точка в конце важна)

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

, конечно, использование firehose влияет на производительность, но в в моем случае это не имеет большого значения, так как rabbitmq недостаточно используется.

Поскольку у меня есть 16 очередей для прослушивания, я бы предпочел не использовать разные шаблоны для каждой очереди. Я мог бы использовать одну очередь ответов для всех очередей RP C, но это было бы узким местом.

поэтому, если нет жесткого NO GO, пожарный шланг кажется хорошим вариантом.

...