Spring Boot应用启动失败时Kafka消费者仍消费的问题咨询

我们有一个Spring Boot应用，使用版本为3.2.6的Spring Cloud Stream Kafka Binder实现Kafka连接。当应用因端口已被占用错误启动失败时，Kafka Consumer仍会开始消费消息，导致消息处理失败并进入DLQ，同时出现“ApplicationContext尚未刷新”的错误。我们使用Spring Boot默认的Tomcat服务器，相关错误栈如下：

应用启动时异常 #

Exception occurred during context initialization port 65000 already in use

处理ContextClosedEvent时抛出的异常 #

Exception thrown from ApplicationListener handling ContextClosedEvent
org.springframework.beans.factory.BeanCreationNotAllowedException: Error creating bean with name 'registrationListener': Singleton bean creation not allowed while singletons of this factory are in destruction (Do not request a bean from a BeanFactory in a destroy method implementation!)
at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:220) ~[spring-beans-5.3.19.jar:5.3.19]
at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:333) ~[spring-beans-5.3.19.jar:5.3.19]
at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:208) ~[spring-beans-5.3.19.jar:5.3.19]
at org.springframework.context.support.AbstractApplicationContext.getBean(AbstractApplicationContext.java:1154) ~[spring-context-5.3.19.jar:5.3.19]
at org.springframework.context.event.ApplicationListenerMethodAdapter.getTargetBean(ApplicationListenerMethodAdapter.java:371) ~[spring-context-5.3.19.jar:5.3.19]
at org.springframework.context.event.ApplicationListenerMethodAdapter.doInvoke(ApplicationListenerMethodAdapter.java:336) ~[spring-context-5.3.19.jar:5.3.19]
at org.springframework.context.event.ApplicationListenerMethodAdapter.processEvent(ApplicationListenerMethodAdapter.java:229) ~[spring-context-5.3.19.jar:5.3.19]
at org.springframework.context.event.ApplicationListenerMethodAdapter.onApplicationEvent(ApplicationListenerMethodAdapter.java:166) ~[spring-context-5.3.19.jar:5.3.19]
at org.springframework.context.event.SimpleApplicationEventMulticaster.doInvokeListener(SimpleApplicationEventMulticaster.java:176) ~[spring-context-5.3.19.jar:5.3.19]
at org.springframework.context.event.SimpleApplicationEventMulticaster.invokeListener(SimpleApplicationEventMulticaster.java:169) ~[spring-context-5.3.19.jar:5.3.19]
at org.springframework.context.event.SimpleApplicationEventMulticaster.multicastEvent(SimpleApplicationEventMulticaster.java:143) ~[spring-context-5.3.19.jar:5.3.19]
at org.springframework.context.support.AbstractApplicationContext.publishEvent(AbstractApplicationContext.java:421) ~[spring-context-5.3.19.jar:5.3.19]
at org.springframework.context.support.AbstractApplicationContext.publishEvent(AbstractApplicationContext.java:427) ~[spring-context-5.3.19.jar:5.3.19]
at org.springframework.context.support.AbstractApplicationContext.publishEvent(AbstractApplicationContext.java:378) ~[spring-context-5.3.19.jar:5.3.19]
at org.springframework.context.support.AbstractApplicationContext.doClose(AbstractApplicationContext.java:1058) ~[spring-context-5.3.19.jar:5.3.19]
at org.springframework.context.support.AbstractApplicationContext.close(AbstractApplicationContext.java:1021) ~[spring-context-5.3.19.jar:5.3.19]
at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184) ~[?:1.8.0_341]
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193) ~[?:1.8.0_341]
at java.util.HashMap$ValueSpliterator.forEachRemaining(HashMap.java:1652) ~[?:1.8.0_341]
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482) ~[?:1.8.0_341]
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472) ~[?:1.8.0_341]
at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:151) ~[?:1.8.0_341]
at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:174) ~[?:1.8.0_341]
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[?:1.8.0_341]
at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418) ~[?:1.8.0_341]
at org.springframework.cloud.stream.binder.DefaultBinderFactory.destroy(DefaultBinderFactory.java:120) ~[spring-cloud-stream-3.2.6.jar:3.2.6]
at org.springframework.beans.factory.support.DisposableBeanAdapter.destroy(DisposableBeanAdapter.java:213) ~[spring-beans-5.3.19.jar:5.3.19]
at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.destroyBean(DefaultSingletonBeanRegistry.java:587) ~[spring-beans-5.3.19.jar:5.3.19]
at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.destroySingleton(DefaultSingletonBeanRegistry.java:559) ~[spring-beans-5.3.19.jar:5.3.19]
at org.springframework.beans.factory.support.DefaultListableBeanFactory.destroySingleton(DefaultListableBeanFactory.java:1163) ~[spring-beans-5.3.19.jar:5.3.19]
at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.destroySingletons(DefaultSingletonBeanRegistry.java:520) ~[spring-beans-5.3.19.jar:5.3.19]
at org.springframework.beans.factory.support.DefaultListableBeanFactory.destroySingletons(DefaultListableBeanFactory.java:1156) ~[spring-beans-5.3.19.jar:5.3.19]
at org.springframework.context.support.AbstractApplicationContext.destroyBeans(AbstractApplicationContext.java:1106) ~[spring-context-5.3.19.jar:5.3.19]
at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:596) ~[spring-context-5.3.19.jar:5.3.19]
at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:145) ~[spring-boot-2.6.7.jar:2.6.7]
at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:745) ~[spring-boot-2.6.7.jar:2.6.7]
at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:420) ~[spring-boot-2.6.7.jar:2.6.7]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:307) ~[spring-boot-2.6.7.jar:2.6.7]
at com.aa.ve.MainApplication.main(MainApplication.java:52)

StreamBridge发送消息时异常 #

Caused by: java.lang.IllegalStateException: org.springframework.boot.web.servlet.context.AnnotationConfigServletWebServerApplicationContext@25da615a has not been refreshed yet
at org.springframework.context.support.AbstractApplicationContext.assertBeanFactoryActive(AbstractApplicationContext.java:1141) ~[spring-context-5.3.19.jar:5.3.19]
at org.springframework.context.support.AbstractApplicationContext.getBean(AbstractApplicationContext.java:1171) ~[spring-context-5.3.19.jar:5.3.19]
at org.springframework.cloud.stream.function.StreamBridge.addInterceptors(StreamBridge.java:325) ~[spring-cloud-stream-3.2.6.jar:3.2.6]
at org.springframework.cloud.stream.function.StreamBridge.resolveDestination(StreamBridge.java:300) ~[spring-cloud-stream-3.2.6.jar:3.2.6]
at org.springframework.cloud.stream.function.StreamBridge.send(StreamBridge.java:217) ~[spring-cloud-stream-3.2.6.jar:3.2.6]
at org.springframework.cloud.stream.function.StreamBridge.send(StreamBridge.java:171) ~[spring-cloud-stream-3.2.6.jar:3.2.6]
at com.de.ve.CustomPublisher.send(CustomPublisher.java:78)

已尝试的解决方案 #

我采取的方案是将消费者的auto-startup属性设置为false：

spring:
  cloud:
    stream:
      bindings:
        consumer-1:
          consumer:
            auto-startup: false

同时编写了实现ApplicationListener的CustomListener，在onApplicationEvent方法中将端点状态改为STARTED：

public class CustomListener implements ApplicationListener<ApplicationReadyEvent> {

    @Autowired
    BindingsEndpoint endpoint; 

    @Override
    public void onApplicationEvent(ApplicationReadyEvent event){
        endpoint.changeState("consumer-1", BindingsLifecycleController.State.STARTED);
    }
}

请问该方案是否合理？还有其他替代方案可以处理此问题吗？

方案合理性判断 #

你的方案完全合理。ApplicationReadyEvent是Spring Boot在ApplicationContext完全刷新、所有Bean初始化完成、Web服务器（如Tomcat）启动成功后才会触发的事件。如果应用因端口占用启动失败，这个事件根本不会触发，消费者也就不会被启动，完美避免了启动过程中消费者提前消费消息导致的问题。

需要注意的是，CustomListener中的BindingsEndpoint必须通过@Autowired或构造函数注入，否则会出现空指针异常，这点在代码中补充后会更严谨。

替代方案 #

1. 基于SmartLifecycle接口自定义启停逻辑 #

实现SmartLifecycle接口，自定义消费者的启动时机，可添加额外校验逻辑：

@Component
public class ConsumerLifecycleManager implements SmartLifecycle {

    @Autowired
    BindingsEndpoint endpoint;
    private boolean isRunning = false;

    @Override
    public boolean isAutoStartup() {
        // 禁止自动启动
        return false;
    }

    @Override
    public void start() {
        // 可在此添加自定义校验逻辑，比如端口可用性检查
        endpoint.changeState("consumer-1", BindingsLifecycleController.State.STARTED);
        isRunning = true;
    }

    @Override
    public void stop() {
        endpoint.changeState("consumer-1", BindingsLifecycleController.State.STOPPED);
        isRunning = false;
    }

    @Override
    public boolean isRunning() {
        return isRunning;
    }
}

同样监听ApplicationReadyEvent触发start()方法，相比直接操作BindingsEndpoint，这种方式扩展性更强。

2. 全局禁用Stream生命周期自动管理 #

通过配置全局禁用Spring Cloud Stream的自动生命周期管理，自行控制所有绑定的启停：

spring:
  cloud:
    stream:
      lifecycle:
        enabled: false

这种方式适合需要统一管理所有消费者/生产者的场景，但需要自行处理所有绑定的生命周期，复杂度较高，不如针对单个消费者设置auto-startup灵活。

3. 升级Spring Cloud Stream版本 #

这个问题本质是旧版本Spring Cloud Stream在ApplicationContext刷新失败时，未正确停止已启动的消费者。升级到3.2.x后续维护版本，或匹配Spring Boot 2.6.x的最新兼容版本，可能已修复该bug，无需自定义代码即可解决。升级前需确认版本兼容性，避免引入其他问题。

4. 监听ContextRefreshedEvent（谨慎使用） #

ContextRefreshedEvent在ApplicationContext刷新完成后触发，但早于ApplicationReadyEvent，此时Web服务器可能还未启动完成。如果仅需确保ApplicationContext刷新完成后启动消费者，可使用该事件，但端口占用时ApplicationContext刷新失败，该事件也不会触发，同样能避免消费者启动。不过对于依赖Web服务就绪的场景，还是ApplicationReadyEvent更合适。

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