如何实现RabbitMQ未启动时Spring集成RabbitMQ Streams应用仍可启动？

针对RabbitMQ Streams的StreamListenerContainer，目前Spring AMQP并没有提供类似普通队列missing-queues-fatal=false的直接配置项，但可以通过以下几种方式实现应用启动时不依赖MQ可用性：

1. 延迟启动StreamListenerContainer并添加重试逻辑 #

• 先将容器的autoStartup设置为false，避免应用启动阶段立即尝试连接MQ：

  @Bean
  public StreamListenerContainer streamListenerContainer(ConnectionFactory connectionFactory) {
      return StreamListenerContainer.create(connectionFactory)
              .autoStartup(false)
              .stream("target-stream")
              .consumer(consumer -> consumer.name("stream-consumer"))
              .build();
  }
  

• 自定义组件监听应用启动事件，在应用就绪后尝试启动容器，并加入重试机制，确保MQ恢复后能自动启动消费：

  @Component
  public class StreamContainerInitializer {
      private final StreamListenerContainer streamContainer;
      private final RetryTemplate retryTemplate;
  
      public StreamContainerInitializer(StreamListenerContainer streamContainer) {
          this.streamContainer = streamContainer;
          // 配置重试策略：最多重试5次，间隔5秒
          SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
          retryPolicy.setMaxAttempts(5);
          FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
          backOffPolicy.setBackOffPeriod(5000);
  
          this.retryTemplate = new RetryTemplate();
          this.retryTemplate.setRetryPolicy(retryPolicy);
          this.retryTemplate.setBackOffPolicy(backOffPolicy);
      }
  
      @EventListener(ApplicationReadyEvent.class)
      public void initStreamContainer() {
          retryTemplate.execute(context -> {
              if (!streamContainer.isRunning()) {
                  streamContainer.start();
              }
              return null;
          });
      }
  }
  

2. 配置连接工厂的自动恢复与重试 #

通过配置RabbitMQ连接工厂的自动恢复机制，让连接失败时自动重试，避免因启动时MQ不可用导致应用崩溃：

@Bean
public CachingConnectionFactory rabbitConnectionFactory() {
    CachingConnectionFactory factory = new CachingConnectionFactory("mq-host");
    factory.setPort(5672);
    factory.setUsername("mq-user");
    factory.setPassword("mq-password");

    // 启用自动连接恢复
    factory.setAutomaticRecoveryEnabled(true);
    // 网络恢复间隔（重试间隔）
    factory.setNetworkRecoveryInterval(5000);
    // 设置连接超时时间
    factory.setConnectionTimeout(3000);
    // 设置心跳检测
    factory.setRequestedHeartBeat(60);

    return factory;
}

这种方式下，即使启动时MQ不可用，连接工厂会持续尝试重连，当MQ恢复后，StreamListenerContainer会自动建立连接并开始消费。

3. 条件化创建StreamListenerContainer（可选） #

如果是开发环境等场景，可以通过配置开关控制容器是否创建，避免启动时的MQ依赖：

@Bean
@ConditionalOnProperty(name = "rabbitmq.streams.enabled", havingValue = "true", matchIfMissing = true)
public StreamListenerContainer streamListenerContainer(ConnectionFactory connectionFactory) {
    return StreamListenerContainer.create(connectionFactory)
            .stream("target-stream")
            .consumer(consumer -> consumer.name("stream-consumer"))
            .build();
}

通过配置文件rabbitmq.streams.enabled=false即可跳过容器创建，实现无MQ时启动应用。

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