Spring Boot非Web应用中Azure EventHubConsumerAsyncClient无法持续运行

我来帮你解决这个问题！你的非Web Spring Boot应用启动后立即退出，核心原因在于非Web环境下Spring Boot没有默认的线程维持进程活跃，加上异步的Event Hub客户端订阅没有阻止应用关闭的机制。下面是具体的分析和解决方案：

你的非Web Spring Boot应用启动后立刻退出，主要有两个核心原因：

1. 缺少进程存活机制：Spring Boot的WebApplicationType.NONE模式下，没有Web容器（比如Tomcat）的线程来维持进程运行。当所有Bean初始化完成后，主线程结束，JVM就会终止。
2. 异步客户端线程属性：你使用的EventHubConsumerAsyncClient基于Reactor实现，其订阅线程默认可能是守护线程——守护线程不会阻止JVM退出，当所有非守护线程结束时，JVM就会关闭。

另外还要注意：你的EventProcessorHostService类如果没有添加@Service或@Component注解，Spring容器不会自动托管这个Bean，@PostConstruct标注的run方法根本不会执行（从你的日志看已经触发了Event Hub连接，应该是你实际代码里加了，但还是提醒一下）。

下面提供两种可靠的解决方式，任选其一即可：

方案一：用CountDownLatch阻塞主线程 #

通过CountDownLatch让主线程等待，直到应用被手动终止（比如Ctrl+C），同时监听上下文关闭事件来释放Latch。

修改你的主类：

@EnableAsync
@SpringBootApplication
public class AzureEventhubConsumerApplication {

    public static void main(String[] args) throws InterruptedException {
        ConfigurableApplicationContext context = new SpringApplicationBuilder(AzureEventhubConsumerApplication.class)
                .web(WebApplicationType.NONE)
                .run(args);
        
        // 阻塞主线程，直到应用关闭
        CountDownLatch shutdownLatch = context.getBean(CountDownLatch.class);
        shutdownLatch.await();
    }

    // 创建一个CountDownLatch实例，用于阻塞主线程
    @Bean
    public CountDownLatch shutdownLatch() {
        return new CountDownLatch(1);
    }

    // 监听应用关闭事件，释放Latch
    @Bean
    public ApplicationListener<ContextClosedEvent> contextClosedListener(CountDownLatch shutdownLatch) {
        return event -> shutdownLatch.countDown();
    }
}

同时确保EventProcessorHostService添加@Service注解：

@Service
public class EventProcessorHostService {
    @Autowired
    EventhubProperties ehProps;
    
    @PostConstruct
    public void run() {
        EventHubConsumerAsyncClient client = new EventHubClientBuilder()
                .connectionString(ehProps.getConnectionString(), ehProps.getEventHubName())
                .consumerGroup(ehProps.getStorage().getConsumerGroupName())
                .buildAsyncConsumerClient();
        
        client.receive(true).subscribe(event -> {
            PartitionContext context = event.getPartitionContext();
            EventData eData = event.getData();
            System.out.printf("Event %s is from partition %s%n.", eData.getSequenceNumber(), context.getPartitionId());
        });
    }
}

方案二：实现SmartLifecycle接口管理生命周期 #

通过SmartLifecycle接口告诉Spring这个Bean是活跃的，需要维持应用运行，同时可以在应用关闭时优雅地清理Event Hub订阅资源。

修改EventProcessorHostService：

@Service
public class EventProcessorHostService implements SmartLifecycle {
    private boolean isRunning = false;
    private Disposable eventSubscription;
    
    @Autowired
    EventhubProperties ehProps;

    @Override
    public void start() {
        // 创建Event Hub异步客户端并订阅消息
        EventHubConsumerAsyncClient client = new EventHubClientBuilder()
                .connectionString(ehProps.getConnectionString(), ehProps.getEventHubName())
                .consumerGroup(ehProps.getStorage().getConsumerGroupName())
                .buildAsyncConsumerClient();
        
        eventSubscription = client.receive(true).subscribe(event -> {
            PartitionContext context = event.getPartitionContext();
            EventData eData = event.getData();
            System.out.printf("Event %s is from partition %s%n.", eData.getSequenceNumber(), context.getPartitionId());
        });
        isRunning = true;
    }

    @Override
    public void stop() {
        // 优雅取消订阅，释放资源
        if (eventSubscription != null) {
            eventSubscription.dispose();
        }
        isRunning = false;
    }

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

    // 以下为默认实现，无需修改
    @Override
    public boolean isAutoStartup() {
        return true;
    }

    @Override
    public void stop(Runnable callback) {
        stop();
        callback.run();
    }

    @Override
    public int getPhase() {
        return 0;
    }
}

额外优化：确保线程为非守护线程 #

如果你的TaskExecutor线程是守护线程，也可能导致JVM提前退出，可以配置线程工厂设置为非守护线程：

@Bean
public TaskExecutor taskexecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(5);
    executor.setMaxPoolSize(10);
    executor.setThreadNamePrefix("eventhub-consumer-");
    
    // 设置线程为非守护线程
    executor.setThreadFactory(runnable -> {
        Thread thread = new Thread(runnable);
        thread.setDaemon(false);
        return thread;
    });
    
    executor.initialize();
    return executor;
}

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