С какими очередями сообщений работал

С какими очередями сообщений работал

Message Queues в Java

Message Queue — это система, которая позволяет асинхронно отправлять и обрабатывать сообщения между компонентами приложения. Я работал с несколькими популярными реализациями.

1. RabbitMQ

Использование

RabbitMQ был основной очередью в микросервисной архитектуре нашего проекта.

Пример с Spring Boot

// Конфигурация
@Configuration
public class RabbitConfig {
    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 DirectExchange exchange() {
        return new DirectExchange(EXCHANGE_NAME, true, false);
    }

    @Bean
    public Queue queue() {
        return new Queue(QUEUE_NAME, true);
    }

    @Bean
    public Binding binding(Queue queue, DirectExchange exchange) {
        return BindingBuilder.bind(queue)
                .to(exchange)
                .with(ROUTING_KEY);
    }
}

// Producer
@Service
public class OrderProducer {
    private final RabbitTemplate rabbitTemplate;

    public void sendOrder(OrderEvent event) {
        rabbitTemplate.convertAndSend(
            RabbitConfig.EXCHANGE_NAME,
            RabbitConfig.ROUTING_KEY,
            event
        );
    }
}

// Consumer
@Service
public class OrderConsumer {
    @RabbitListener(queues = RabbitConfig.QUEUE_NAME)
    public void handleOrder(OrderEvent event) {
        // Обработка заказа
        orderService.process(event);
    }
}

Сложность: Dead Letter Queue

// Если обработка не удалась, сообщение идёт в DLQ
@Configuration
public class RabbitDLQConfig {
    public static final String DLQ_NAME = "order.dlq";
    public static final String DLX_EXCHANGE = "order.dlx";

    @Bean
    public Queue deadLetterQueue() {
        return new Queue(DLQ_NAME);
    }

    @Bean
    public DirectExchange dlxExchange() {
        return new DirectExchange(DLX_EXCHANGE);
    }

    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(deadLetterQueue())
                .to(dlxExchange())
                .with(DLQ_NAME);
    }
}

// Main Queue с DLX параметрами
@Bean
public Queue mainQueue() {
    return QueueBuilder.durable(QUEUE_NAME)
            .deadLetterExchange(DLX_EXCHANGE)
            .deadLetterRoutingKey(DLQ_NAME)
            .ttl(60000)  // TTL 1 минута
            .maxLength(10000)  // Max 10k сообщений
            .build();
}

Сценарий: если сервис недоступен, сообщение автоматически отправляется в DLQ для повторной обработки.

2. Apache Kafka

Использование

Kafka использовалась для high-throughput систем (логирование, аналитика, event streaming).

Пример

// Конфигурация 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);
        configProps.put(ProducerConfig.ACKS_CONFIG, "all");  // Все реплики должны подтвердить
        configProps.put(ProducerConfig.RETRIES_CONFIG, 3);
        return new DefaultProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, OrderEvent> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

// Producer
@Service
public class OrderKafkaProducer {
    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;

    public void sendOrder(OrderEvent event) {
        kafkaTemplate.send("order-topic", event.getOrderId(), event)
                .addCallback(
                    result -> logger.info("Sent: " + event),
                    ex -> logger.error("Failed: " + ex.getMessage())
                );
    }
}

// Consumer
@Service
public class OrderKafkaConsumer {
    @KafkaListener(topics = "order-topic", groupId = "order-group")
    public void consume(OrderEvent event) {
        orderService.process(event);
    }
}

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

• Масштабируемость: обрабатывает миллионы сообщений/сек
• Persistence: сообщения хранятся, можно переворачивать
• Consumer Groups: несколько потребителей могут читать параллельно
• Exactly-Once Semantics: гарантия единожды обработки

3. AWS SQS (Simple Queue Service)

Использование

SQS для облачных приложений на AWS.

Пример

// Зависимость
// <dependency>
//     <groupId>software.amazon.awssdk</groupId>
//     <artifactId>sqs</artifactId>
// </dependency>

@Service
public class SqsMessageProducer {
    private final SqsClient sqsClient;
    private final String queueUrl;

    public void send(String message) {
        SendMessageRequest request = SendMessageRequest.builder()
                .queueUrl(queueUrl)
                .messageBody(message)
                .build();
        
        sqsClient.sendMessage(request);
    }
}

// Consumer
@Service
@Scheduled(fixedDelay = 1000)
public class SqsMessageConsumer {
    private final SqsClient sqsClient;
    private final String queueUrl;

    public void consume() {
        ReceiveMessageRequest request = ReceiveMessageRequest.builder()
                .queueUrl(queueUrl)
                .maxNumberOfMessages(10)
                .waitTimeSeconds(20)  // Long polling
                .build();

        List<Message> messages = sqsClient.receiveMessage(request).messages();
        
        for (Message msg : messages) {
            processMessage(msg);
            // Удаляем сообщение после успешной обработки
            deleteMessage(msg.receiptHandle());
        }
    }
}

4. Сравнение всех трёх

                RabbitMQ          Kafka              SQS
Производ.     10k-100k msg/s    1M+ msg/s         Очень высокая
Латенция      Низкая            Средняя           Средняя
Персистент.   Да                Да                Да
Жёсткость    Жёсткая           Гибкая             Гибкая
Услуга       Self-hosted        Self-hosted        Managed
Сложность    Средняя            Сложная            Простая
Usage        Микросервисы       Event Streaming    AWS приложения

Сложные задачи с Message Queues

1. Именно один раз обработка (Idempotency)

@Service
public class IdempotentOrderProcessor {
    private final RedisTemplate<String, Object> redis;
    private final OrderService orderService;

    public void processOrder(OrderEvent event) {
        String key = "order:" + event.getOrderId();
        
        // Проверяем, не обработали ли мы это уже
        if (redis.hasKey(key)) {
            logger.info("Order already processed: " + event.getOrderId());
            return;
        }
        
        // Обрабатываем
        orderService.save(event);
        
        // Помечаем как обработанное
        redis.opsForValue().set(key, "processed", Duration.ofHours(24));
    }
}

2. Retry с экспоненциальной задержкой

@Service
public class ResilientOrderProcessor {
    
    @Retryable(
        value = {Exception.class},
        maxAttempts = 3,
        backoff = @Backoff(delay = 1000, multiplier = 2)
    )
    public void processOrder(OrderEvent event) {
        // 1я попытка: 1 сек
        // 2я попытка: 2 сек
        // 3я попытка: 4 сек
        // Если все не удалась — exception
        orderService.process(event);
    }
}

На собеседовании

Хороший ответ:

"Я работал с RabbitMQ, Kafka и AWS SQS в разных контекстах. RabbitMQ использовал для синхронизации микросервисов с гарантией доставки и Dead Letter Queue обработкой. Kafka применял для high-throughput систем с логированием и аналитикой. SQS использовал в AWS приложениях.

Самая сложная задача была реализовать exactly-once семантику — используя Redis для идемпотентности и обработку DLQ сообщений. Это гарантирует, что платежи не обрабатываются дважды.

Понимаю trade-offs между ними: RabbitMQ проще для стартапа, Kafka масштабируется лучше, SQS удобнее в AWS экосистеме."