Spring Cloud Stream中RabbitMQ Broker宕机检测与非重试消息推送问询

Hey there! Let's break down how to handle RabbitMQ downtime in Spring Cloud Stream without relying on retry mechanisms. The key here is to actively detect the broker's status and adjust your message publishing logic based on that check. Here's a step-by-step approach:

Instead of polling the broker constantly, leverage RabbitMQ's connection events to maintain a real-time view of its availability. Spring AMQP publishes events like ConnectionEstablishedEvent and ConnectionClosedEvent when the broker connection changes state.

Create a listener component to track the status:

@Component
public class RabbitBrokerStatusMonitor {
    // Volatile ensures visibility across threads
    private volatile boolean isBrokerAvailable = false;

    @EventListener
    public void onConnectionEstablished(ConnectionEstablishedEvent event) {
        isBrokerAvailable = true;
        log.info("RabbitMQ connection established - broker is now available");
    }

    @EventListener
    public void onConnectionClosed(ConnectionClosedEvent event) {
        isBrokerAvailable = false;
        log.warn("RabbitMQ connection closed - broker is unavailable");
    }

    public boolean isBrokerAvailable() {
        return isBrokerAvailable;
    }
}

Now, inject this status monitor into your message publishing service to gate message sends. Using StreamBridge (the recommended way to send messages in Spring Cloud Stream 3.x+), we'll only publish if the broker is confirmed available:

@Service
public class StreamMessagePublisher {
    private final StreamBridge streamBridge;
    private final RabbitBrokerStatusMonitor statusMonitor;
    private final OutgoingMessageRepository messageRepository;

    public StreamMessagePublisher(StreamBridge streamBridge,
                                 RabbitBrokerStatusMonitor statusMonitor,
                                 OutgoingMessageRepository messageRepository) {
        this.streamBridge = streamBridge;
        this.statusMonitor = statusMonitor;
        this.messageRepository = messageRepository;
    }

    public void publishMessage(String bindingTarget, Object payload) {
        if (statusMonitor.isBrokerAvailable()) {
            // Send immediately if broker is up
            streamBridge.send(bindingTarget, payload);
            log.debug("Message sent to {}: {}", bindingTarget, payload);
        } else {
            // Persist message for later delivery when broker recovers
            OutgoingMessage unsentMessage = new OutgoingMessage();
            unsentMessage.setBindingTarget(bindingTarget);
            unsentMessage.setPayload(payload.toString());
            unsentMessage.setCreatedAt(LocalDateTime.now());
            messageRepository.save(unsentMessage);
            log.warn("Broker unavailable - persisted message for later delivery: {}", payload);
        }
    }
}

When the broker comes back online, we can trigger a resend of all unsent messages by extending our connection listener:

// Add this to RabbitBrokerStatusMonitor
@Autowired
private StreamBridge streamBridge;
@Autowired
private OutgoingMessageRepository messageRepository;

@EventListener
public void onConnectionEstablished(ConnectionEstablishedEvent event) {
    isBrokerAvailable = true;
    log.info("RabbitMQ connection restored - starting resend of unsent messages");
    
    // Fetch all unsent messages (adjust query based on your repo setup)
    List<OutgoingMessage> unsentMessages = messageRepository.findBySentFalse();
    
    for (OutgoingMessage msg : unsentMessages) {
        try {
            streamBridge.send(msg.getBindingTarget(), msg.getPayload());
            msg.setSent(true);
            msg.setSentAt(LocalDateTime.now());
            messageRepository.save(msg);
        } catch (Exception e) {
            log.error("Failed to resend message {}: {}", msg.getId(), e.getMessage());
            // Optionally mark as failed or retry later manually
        }
    }
}

For edge cases where event listeners might miss state changes (e.g., network blips that don't trigger a full connection close), you can add a manual check using the RabbitMQ connection factory:

// Add this to RabbitBrokerStatusMonitor
@Autowired
private CachingConnectionFactory connectionFactory;

public boolean performManualHealthCheck() {
    try (Connection connection = connectionFactory.createConnection()) {
        return connection.isOpen();
    } catch (AmqpException e) {
        log.debug("Manual health check failed: {}", e.getMessage());
        return false;
    }
}

You can call this method periodically (using @Scheduled) to refresh the status, or use it as a fallback in your publishing logic.

• Thread Safety: Use volatile for the status flag to ensure all threads see the latest state.
• Persistence: Choose a reliable storage (like a relational database) for unsent messages to avoid data loss.
• Error Handling: Add retries for resending messages (but only after the broker is back up) or implement a dead-letter queue for messages that repeatedly fail.
• Configuration: Tune spring.rabbitmq.connection-timeout to a reasonable value (e.g., 5000ms) to avoid long waits during manual checks.

内容的提问来源于stack exchange，提问作者Prash