AWS FirehoseAsync主函数结束后如何停止线程池，避免主进程持续运行？

嗨，我碰到过类似的问题，这其实是AWS异步客户端的线程池在搞鬼！让我给你拆解下原因和解决办法：

问题根源 #

你创建的AmazonKinesisFirehoseAsync客户端会使用自定义的ThreadPoolExecutorFactory生成线程池，而默认情况下线程池里的线程是非守护线程。这类线程不会随着主线程的结束而自动终止，它们会一直处于等待新任务的状态，导致JVM无法正常退出，主进程就僵住了。另外，你的代码在提交完所有异步任务后，没有主动关闭Firehose客户端，也没有处理线程池的收尾逻辑。

解决方案 #

这里有几个靠谱的解决方式，你可以根据自己的场景选择：

1. 显式关闭Firehose异步客户端 #

在所有异步任务提交完成后，主动调用客户端的关闭方法，让线程池优雅终止。这样能保证所有已提交的任务都处理完毕再退出：

// 提交完所有10条日志任务后
for(int i=0; i<10; i++) {
    // ... 你的日志提交代码 ...
}

// 开始关闭客户端
kinesisFirehoseAsync.shutdown();
try {
    // 等待10秒让线程池处理完剩余任务
    if (!kinesisFirehoseAsync.awaitTermination(10, TimeUnit.SECONDS)) {
        // 如果超时就强制关闭
        kinesisFirehoseAsync.shutdownNow();
    }
} catch (InterruptedException e) {
    // 被中断时也强制关闭，并恢复中断状态
    kinesisFirehoseAsync.shutdownNow();
    Thread.currentThread().interrupt();
}

2. 把线程池的线程设置为守护线程 #

如果你希望主线程结束后，异步线程自动跟着终止（要注意可能会丢失未完成的任务），可以修改ThreadPoolExecutorFactory，让它创建守护线程：

public class ThreadPoolExecutorFactory implements ExecutorFactory {
    private final int corePoolSize;
    private final BlockingQueue<Runnable> taskBuffer;

    public ThreadPoolExecutorFactory(int corePoolSize, BlockingQueue<Runnable> taskBuffer) {
        this.corePoolSize = corePoolSize;
        this.taskBuffer = taskBuffer;
    }

    @Override
    public ExecutorService newExecutor() {
        return new ThreadPoolExecutor(corePoolSize, corePoolSize,
                0L, TimeUnit.MILLISECONDS,
                taskBuffer,
                // 自定义ThreadFactory，创建守护线程
                r -> {
                    Thread thread = new Thread(r);
                    thread.setDaemon(true); // 关键：设置为守护线程
                    thread.setName("firehose-async-worker"); // 给线程起个名字方便排查问题
                    return thread;
                });
    }
}

这样主线程执行完后，JVM会自动终止所有守护线程，主进程就能正常退出了。

3. 结合Log4j Appender的生命周期管理 #

因为你是把Firehose作为Log4j的Appender使用，最好在Appender的停止方法里处理客户端关闭逻辑，这样当Log4j关闭时（比如应用程序退出时），自动清理资源：

public class FirehoseLog4jAppender extends AppenderSkeleton {
    private AmazonKinesisFirehoseAsync kinesisFirehoseAsync;

    // ... 其他初始化代码 ...

    @Override
    public void stop() {
        super.stop();
        if (kinesisFirehoseAsync != null) {
            kinesisFirehoseAsync.shutdown();
            try {
                // 等待5秒处理剩余任务
                kinesisFirehoseAsync.awaitTermination(5, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

注意事项 #

• 用守护线程的方式要谨慎，尤其是生产环境：如果主线程提前退出，可能会导致部分日志还没发送到Firehose就被中断，造成日志丢失。
• 显式关闭客户端是更稳妥的生产环境做法，能确保所有提交的日志任务都被处理完毕。

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