RabbitMQ mit Spring Boot: Asynchrone Kommunikation in Microservices

Das Problem mit synchroner Kommunikation

Konzepte: Exchange, Queue, Binding

• Exchange: Nimmt Nachrichten entgegen und entscheidet, in welche Queue sie kommen
• Queue: Speichert Nachrichten bis ein Consumer sie abholt
• Binding: Verbindet Exchange mit Queue (über Routing Key)
• Routing Key: String-Schlüssel zur Weiterleitung (z. B. order.created)

Konfiguration: Exchanges & Queues

@Configuration
public class RabbitMQConfig {

    public static final String ORDER_EXCHANGE    = "order.exchange";
    public static final String ORDER_CREATED_QUEUE = "order.created.queue";
    public static final String ORDER_CREATED_KEY   = "order.created";

    @Bean
    public TopicExchange orderExchange() {
        return new TopicExchange(ORDER_EXCHANGE, true, false);
    }

    @Bean
    public Queue orderCreatedQueue() {
        return QueueBuilder.durable(ORDER_CREATED_QUEUE)
                .withArgument("x-dead-letter-exchange", "")
                .withArgument("x-dead-letter-routing-key", "dead.letter.queue")
                .build();
    }

    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
}

Producer: Nachrichten senden

@Service
@RequiredArgsConstructor
public class OrderEventPublisher {

    private final RabbitTemplate rabbitTemplate;

    public void publishOrderCreated(Order order) {
        var event = new OrderCreatedEvent(
            order.getId(), order.getCustomerId(),
            order.getTotalAmount(), Instant.now()
        );
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.ORDER_EXCHANGE,
            RabbitMQConfig.ORDER_CREATED_KEY,
            event
        );
    }
}

// Wichtig: Event erst NACH dem DB-Commit senden!
@Transactional
public OrderResponse placeOrder(OrderRequest request) {
    Order saved = orderRepository.save(buildOrder(request));
    TransactionSynchronizationManager.registerSynchronization(
        new TransactionSynchronizationAdapter() {
            @Override public void afterCommit() {
                eventPublisher.publishOrderCreated(saved);
            }
        }
    );
    return toResponse(saved);
}

Consumer: Nachrichten empfangen

@Component
@RequiredArgsConstructor
public class OrderCreatedListener {

    @RabbitListener(queues = RabbitMQConfig.ORDER_CREATED_QUEUE)
    public void handleOrderCreated(
            OrderCreatedEvent event,
            Channel channel,
            @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag
    ) {
        try {
            paymentService.processPayment(event);
            channel.basicAck(deliveryTag, false);   // Erfolg

        } catch (PaymentException e) {
            // Nicht-retry-fähig → Dead Letter Queue
            channel.basicNack(deliveryTag, false, false);

        } catch (Exception e) {
            // Retry → zurück in Queue
            channel.basicNack(deliveryTag, false, true);
        }
    }
}

Fazit

• Entkopplung: Services kennen sich nicht gegenseitig
• Fehlertoleranz: Nachrichten überleben Service-Ausfälle
• Dead Letter Queue: Kein Datenverlust bei Verarbeitungsfehlern
• Skalierbarkeit: Mehrere Consumer-Instanzen verarbeiten parallel
• Transaktionssicherheit: Event erst nach DB-Commit senden