Что такое RabbitMQ?

Что такое RabbitMQ?

RabbitMQ - это открытый брокер сообщений, реализующий протокол AMQP (Advanced Message Queuing Protocol). Он предназначен для асинхронной коммуникации между различными частями распределённого приложения. RabbitMQ позволяет системам обмениваться сообщениями, не блокируя друг друга.

Основные концепции

Broker: центральный сервер, который получает и доставляет сообщения.

Publisher (Producer): приложение, которое отправляет сообщения в RabbitMQ.

Subscriber (Consumer): приложение, которое получает и обрабатывает сообщения из RabbitMQ.

Queue: очередь, где хранятся сообщения до того, как их заберёт consumer.

Exchange: компонент, который определяет, как маршрутизировать сообщения из publisher в queue.

Binding: связь между exchange и queue, определяющая правила маршрутизации.

Типы Exchange

Direct Exchange: маршрутизирует сообщение в queue, где routing key точно совпадает с binding key.

Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();

channel.exchangeDeclare("direct_exchange", "direct");
channel.queueDeclare("direct_queue", false, false, false, null);
channel.queueBind("direct_queue", "direct_exchange", "order.created");

channel.basicPublish("direct_exchange", "order.created", null, "Order123".getBytes());

Fanout Exchange: отправляет сообщение во все связанные queue без учета routing key.

channel.exchangeDeclare("fanout_exchange", "fanout");
channel.queueDeclare("notification_queue", false, false, false, null);
channel.queueDeclare("analytics_queue", false, false, false, null);

channel.queueBind("notification_queue", "fanout_exchange", "");
channel.queueBind("analytics_queue", "fanout_exchange", "");

channel.basicPublish("fanout_exchange", "", null, "broadcast_message".getBytes());

Topic Exchange: маршрутизирует сообщения на основе шаблона routing key.

channel.exchangeDeclare("topic_exchange", "topic");
channel.queueDeclare("user_events_queue", false, false, false, null);

channel.queueBind("user_events_queue", "topic_exchange", "user.*");
// Ловит: user.created, user.updated, user.deleted

channel.basicPublish("topic_exchange", "user.created", null, "user_data".getBytes());

Практический пример с Spring Boot

@Configuration
public class RabbitMQConfig {
    
    public static final String EXCHANGE_NAME = "order_exchange";
    public static final String QUEUE_NAME = "order_queue";
    public static final String ROUTING_KEY = "order.#";
    
    @Bean
    public TopicExchange exchange() {
        return new TopicExchange(EXCHANGE_NAME, true, false);
    }
    
    @Bean
    public Queue queue() {
        return new Queue(QUEUE_NAME, true);
    }
    
    @Bean
    public Binding binding(Queue queue, TopicExchange exchange) {
        return BindingBuilder.bind(queue)
            .to(exchange)
            .with(ROUTING_KEY);
    }
}

// Producer - отправляет сообщения
@Service
public class OrderProducer {
    
    private final RabbitTemplate rabbitTemplate;
    
    public OrderProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }
    
    public void sendOrderCreated(Order order) {
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.EXCHANGE_NAME,
            "order.created",
            order
        );
    }
    
    public void sendOrderCancelled(Order order) {
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.EXCHANGE_NAME,
            "order.cancelled",
            order
        );
    }
}

// Consumer - получает и обрабатывает сообщения
@Service
public class OrderConsumer {
    
    @RabbitListener(queues = RabbitMQConfig.QUEUE_NAME)
    public void handleOrderCreated(Order order) {
        System.out.println("Processing order: " + order.getId());
        // Логика обработки заказа
        // Отправка уведомления пользователю
        // Запись в БД
    }
}

// Spring Boot зависимость
// pom.xml:
// <dependency>
//     <groupId>org.springframework.boot</groupId>
//     <artifactId>spring-boot-starter-amqp</artifactId>
// </dependency>

// application.yml:
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest

Преимущества RabbitMQ

Асинхронность: publisher не ждёт обработки сообщения consumer, может продолжать работу.

Надежность: сообщения сохраняются в queue до обработки, даже если consumer временно недоступен.

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

Развязка компонентов: publisher не знает о consumer и наоборот, что упрощает разработку.

Гибкость маршрутизации: разные типы exchange позволяют реализовать сложные сценарии доставки.

Недостатки и вызовы

Message ordering: при наличии нескольких consumer порядок обработки может отличаться от порядка отправки.

Duplicate processing: в случае сбоев один consumer может обработать сообщение дважды.

Operational complexity: требует запуска и поддержки дополнительного сервиса.

Практические сценарии использования

Event-driven architecture: отправка событий при создании/обновлении ресурсов.

Микросервисная архитектура: коммуникация между сервисами через очереди.

Background jobs: обработка долгих операций асинхронно.

Rate limiting: очередь может сбалансировать нагрузку на систему.

// Пример обработки долгой операции
@Service
public class ReportService {
    
    private final RabbitTemplate rabbitTemplate;
    
    public void generateReport(String reportId) {
        // Отправляем задачу в очередь
        rabbitTemplate.convertAndSend("report_exchange", "report.generate", reportId);
        // Сразу возвращаем ответ пользователю
    }
}

@Service
public class ReportWorker {
    
    @RabbitListener(queues = "report_queue")
    public void processReport(String reportId) {
        // Долгая операция, может занять минуты
        // Обработка в фоне, не блокирует основной поток
    }
}

Альтернативы

Apache Kafka: более мощный для обработки высоких нагрузок потоков данных.

Redis Streams: проще в настройке, хорош для простых случаев.

Amazon SQS: managed сервис в облаке, no operations needed.

RabbitMQ - это проверенное временем решение для организации асинхронной коммуникации в распределённых системах, особенно популярно в микросервисной архитектуре.