Какие знаешь брокеры сообщений, которые используются в микросервисной архитектуре?

Брокеры сообщений в микросервисной архитектуре

Брокеры сообщений — критически важная компонента микросервисной архитектуры, обеспечивающая асинхронную коммуникацию между сервисами и гарантирующая надёжность доставки. Рассмотрю наиболее популярные решения.

RabbitMQ — message broker на основе AMQP

Рабитемью — один из самых стабильных и проверенных брокеров с более чем 15-летней историей.

Архитектура:

• Exchange — точка входа, распределяет сообщения по очередям
• Queue — хранилище сообщений
• Binding — связь между Exchange и Queue

// Конфигурация
@Configuration
public class RabbitConfig {
    // Объявляем exchange
    @Bean
    public DirectExchange orderExchange() {
        return new DirectExchange("order.exchange", true, false);
    }
    
    // Очередь для обработки платежей
    @Bean
    public Queue paymentQueue() {
        return new Queue("payment.queue", true);
    }
    
    // Binding
    @Bean
    public Binding paymentBinding(Queue paymentQueue, DirectExchange orderExchange) {
        return BindingBuilder.bind(paymentQueue)
            .to(orderExchange)
            .with("payment.*");
    }
}

// Отправка
@Service
public class OrderService {
    @Autowired private RabbitTemplate rabbitTemplate;
    
    public void processOrder(Order order) {
        rabbitTemplate.convertAndSend(
            "order.exchange",
            "payment.process",
            new PaymentEvent(order.getId(), order.getAmount())
        );
    }
}

// Получение
@Component
public class PaymentListener {
    @RabbitListener(queues = "payment.queue")
    public void handlePayment(PaymentEvent event) {
        // Обработка платежа
        logger.info("Processing payment for order: " + event.getOrderId());
    }
}

Плюсы: надёжность, ACK mechanism, priority queues, dead-letter exchanges для ошибок, есть Web UI. Минусы: нужна отдельная инфраструктура, может быть bottleneck для очень высоких нагрузок.

Apache Kafka — distributed event streaming platform

Кафка — разработана LinkedIn для обработки миллиардов событий в день. Основана на идее event log.

Архитектура:

• Topic — логический канал
• Partition — физическое разбиение данных для параллелизма
• Consumer Group — группа потребителей, читающих одну партицию
• Offset — позиция в логе

// Конфигурация Producer
@Configuration
public class KafkaProducerConfig {
    @Bean
    public ProducerFactory<String, OrderEvent> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        return new DefaultProducerFactory<>(configProps);
    }
    
    @Bean
    public KafkaTemplate<String, OrderEvent> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

// Отправка события
@Service
public class OrderService {
    @Autowired private KafkaTemplate<String, OrderEvent> kafkaTemplate;
    
    public void createOrder(Order order) {
        orderRepository.save(order);
        
        kafkaTemplate.send("orders.topic", 
            order.getId().toString(),
            new OrderEvent(order.getId(), order.getStatus())
        );
    }
}

// Consumer
@Configuration
public class KafkaConsumerConfig {
    @Bean
    public ConsumerFactory<String, OrderEvent> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processing-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, OrderEvent.class.getName());
        return new DefaultConsumerFactory<>(props);
    }
}

@Component
public class OrderEventListener {
    @KafkaListener(topics = "orders.topic", groupId = "order-processing-group")
    public void handleOrderEvent(OrderEvent event) {
        logger.info("Received order event: " + event.getOrderId());
    }
}

Плюсы: высокая производительность (миллионы сообщений/сек), durability (сохраняет историю), stream processing, распределённость, replay events. Минусы: больше ресурсов, сложнее конфигурация, оверкилл для простых случаев.

Apache ActiveMQ — универсальный message broker

Этот брокер поддерживает как JMS (Java Messaging Service), так и AMQP, OpenWire протоколы.

@Configuration
public class ActiveMQConfig {
    @Bean
    public JmsTemplate jmsTemplate(ConnectionFactory connectionFactory) {
        return new JmsTemplate(connectionFactory);
    }
}

@Service
public class NotificationService {
    @Autowired private JmsTemplate jmsTemplate;
    
    public void sendNotification(String userId, String message) {
        jmsTemplate.convertAndSend("notifications.queue", 
            new Notification(userId, message));
    }
}

@Component
public class NotificationListener {
    @JmsListener(destination = "notifications.queue")
    public void receiveNotification(Notification notification) {
        // Отправка email/SMS/push
    }
}

Плюсы: поддержка JMS стандарта, несколько протоколов одновременно, хорошая документация. Минусы: не такой популярный как Kafka в новых проектах, может быть медленнее.

AWS SQS / SNS — управляемые облачные сервисы

Для приложений на AWS.

@Configuration
public class SqsConfig {
    @Bean
    public SqsTemplate sqsTemplate(SqsAsyncClient sqsAsyncClient) {
        return SqsTemplate.builder()
            .sqsAsyncClient(sqsAsyncClient)
            .build();
    }
}

@Service
public class OrderService {
    @Autowired private SqsTemplate sqsTemplate;
    
    public void publishOrderEvent(Order order) {
        sqsTemplate.send(to -> to
            .queue("orders-queue")
            .payload(new OrderEvent(order.getId(), order.getStatus()))
        );
    }
}

@Component
public class OrderProcessor {
    @SqsListener("orders-queue")
    public void processOrder(OrderEvent event) {
        logger.info("Processing order: " + event.getOrderId());
    }
}

Плюсы: managed service (не нужна инфраструктура), масштабируется автоматически, дешево для низких объёмов. Минусы: lock-in, высокие затраты при большом трафике.

Redis Streams — для более простых случаев

Когда нужна лёгкая асинхронная коммуникация без heavy infrastructure:

@Service
public class RedisStreamService {
    @Autowired private StringRedisTemplate redisTemplate;
    
    public void publishEvent(String streamKey, Map<String, String> event) {
        redisTemplate.opsForStream().add(
            streamKey,
            event
        );
    }
    
    public void consumeStream(String streamKey) {
        // Consumer group
        redisTemplate.opsForStream().read(
            Consumer.from("my-group", "consumer1"),
            StreamOffset.fromStart(streamKey)
        );
    }
}

Google Pub/Sub — облачная альтернатива Kafka

Для Google Cloud Platform:

@Service
public class PubSubService {
    @Autowired private PubSubTemplate pubSubTemplate;
    
    public void publishMessage(String topic, String message) {
        pubSubTemplate.publish(topic, message);
    }
}

@Component
public class MessageListener {
    @PubSubListener("order-topic")
    public void receiveMessage(String message) {
        logger.info("Received: " + message);
    }
}

Сравнительная таблица

Best practices

1. Идемпотентность — обработчик должен корректно работать при повторной доставке:

@Service
public class PaymentService {
    public void processPayment(PaymentEvent event) {
        // Проверяем, не обработали ли уже
        if (paymentRepository.existsByEventId(event.getEventId())) {
            return;
        }
        // Обработка
        paymentRepository.save(new Payment(event));
    }
}

1. Circuit breaker — обработка неудачных сообщений:

@Service
public class ResilientOrderService {
    @CircuitBreaker(name = "orderService", fallbackMethod = "fallback")
    public void processOrder(Order order) {
        // Обработка
    }
    
    public void fallback(Order order, Exception e) {
        logger.error("Failed to process order", e);
        // Отправить в dead letter queue
    }
}

1. Dead Letter Queue — для сообщений, которые не удалось обработать.

Выбор брокера — стратегическое решение. Для стартапов часто достаточно Redis, для высоконагруженных систем обычно выбирают Kafka.