«1. Введение

В этом руководстве мы рассмотрим концепцию разветвления и обмена темами с помощью Spring AMQP и RabbitMQ.

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

Для этого руководства рекомендуется предварительно прочитать статью Обмен сообщениями с помощью Spring AMQP.

2. Настройка разветвленного обмена

Давайте настроим разветвленный обмен с двумя привязанными к нему очередями. Когда мы отправим сообщение на этот обмен, обе очереди получат сообщение. Наш разветвленный обмен игнорирует любой ключ маршрутизации, включенный в сообщение.

Spring AMQP позволяет нам агрегировать все объявления очередей, обменов и привязок в объекте Declarables: 

@Bean
public Declarables fanoutBindings() {
    Queue fanoutQueue1 = new Queue("fanout.queue1", false);
    Queue fanoutQueue2 = new Queue("fanout.queue2", false);
    FanoutExchange fanoutExchange = new FanoutExchange("fanout.exchange");

    return new Declarables(
      fanoutQueue1,
      fanoutQueue2,
      fanoutExchange,
      bind(fanoutQueue1).to(fanoutExchange),
      BindingBuilder.bind(fanoutQueue2).to(fanoutExchange));
}

3. Настройка темы обмена

Теперь мы также настроим тему exchange с двумя очередями, каждая со своим шаблоном привязки: 

@Bean
public Declarables topicBindings() {
    Queue topicQueue1 = new Queue(topicQueue1Name, false);
    Queue topicQueue2 = new Queue(topicQueue2Name, false);

    TopicExchange topicExchange = new TopicExchange(topicExchangeName);

    return new Declarables(
      topicQueue1,
      topicQueue2,
      topicExchange,
      BindingBuilder
        .bind(topicQueue1)
        .to(topicExchange).with("*.important.*"),
      BindingBuilder
        .bind(topicQueue2)
        .to(topicExchange).with("#.error"));
}

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

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

Наши шаблоны связывания могут использовать звездочку (*) для соответствия слову в определенной позиции или знак решетки (#) для соответствия нулю или более словам.

Итак, наша тема Очередь1 будет получать сообщения с ключами маршрутизации, имеющими шаблон из трех слов со средним словом «важный», например: «user.important.error» или «blog.important». уведомление».

И наша темаQueue2 будет получать сообщения, ключи маршрутизации которых заканчиваются на слово error; примерами соответствия являются «error», «user.important.error» или «blog.post.save.error».

4. Настройка источника

Мы будем использовать метод convertAndSend шаблона RabbitTemplate для отправки примеров сообщений: 

    String message = " payload is broadcast";
    return args -> {
        rabbitTemplate.convertAndSend(FANOUT_EXCHANGE_NAME, "", "fanout" + message);
        rabbitTemplate.convertAndSend(TOPIC_EXCHANGE_NAME, ROUTING_KEY_USER_IMPORTANT_WARN, 
            "topic important warn" + message);
        rabbitTemplate.convertAndSend(TOPIC_EXCHANGE_NAME, ROUTING_KEY_USER_IMPORTANT_ERROR, 
            "topic important error" + message);
    };

Шаблон RabbitTemplate предоставляет множество перегруженных методов convertAndSend() для различных типов обмена.

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

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

5. Настройка потребителей

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

    @RabbitListener(queues = {FANOUT_QUEUE_1_NAME})
    public void receiveMessageFromFanout1(String message) {
        System.out.println("Received fanout 1 message: " + message);
    }

    @RabbitListener(queues = {FANOUT_QUEUE_2_NAME})
    public void receiveMessageFromFanout2(String message) {
        System.out.println("Received fanout 2 message: " + message);
    }

    @RabbitListener(queues = {TOPIC_QUEUE_1_NAME})
    public void receiveMessageFromTopic1(String message) {
        System.out.println("Received topic 1 (" + BINDING_PATTERN_IMPORTANT + ") message: " + message);
    }

    @RabbitListener(queues = {TOPIC_QUEUE_2_NAME})
    public void receiveMessageFromTopic2(String message) {
        System.out.println("Received topic 2 (" + BINDING_PATTERN_ERROR + ") message: " + message);
    }

Мы настроим потребителей, используя аннотацию @RabbitListener. Здесь передается только один аргумент — имя очереди. Потребители здесь не знают об биржах или ключах маршрутизации.

6. Запуск примера

Наш пример проекта представляет собой приложение Spring Boot, поэтому он инициализирует приложение вместе с подключением к RabbitMQ и настроит все очереди, обмены и привязки.

По умолчанию наше приложение ожидает экземпляр RabbitMQ, работающий на локальном хосте через порт 5672. Мы можем изменить это и другие значения по умолчанию в application.yaml.

Наш проект предоставляет конечную точку HTTP в URI — /broadcast — которая принимает POST-сообщения с сообщением в теле запроса.

Когда мы отправляем запрос на этот URI с телом «Test», мы должны увидеть в выводе что-то похожее на это: 

Received fanout 1 message: fanout payload is broadcast
Received topic 1 (*.important.*) message: topic important warn payload is broadcast
Received topic 2 (#.error) message: topic important error payload is broadcast
Received fanout 2 message: fanout payload is broadcast
Received topic 1 (*.important.*) message: topic important error payload is broadcast

Порядок, в котором мы увидим эти сообщения, конечно, не гарантирован. .

7. Заключение

В этом кратком руководстве мы рассмотрели разветвление и обмен темами с помощью Spring AMQP и RabbitMQ.

Полный исходный код и все фрагменты кода для этого руководства доступны в репозитории GitHub.