firebase-admin 9.2.0：sendEachForMulticastAsync线程池未释放，自定义ThreadManager无回调

问题排查与修复步骤 #

1. 确保CustomThreadManager实现正确 #

自定义的ThreadManager必须正确实现接口方法，且合理配置线程池参数，尤其要注意线程是否为守护线程（守护线程可能导致回调未执行完就被终止）。示例实现：

public class CustomThreadManager implements ThreadManager {
    private final ExecutorService fixedThreadPool;

    public CustomThreadManager(int poolSize) {
        // 创建固定大小线程池，设置非守护线程
        this.fixedThreadPool = Executors.newFixedThreadPool(poolSize, runnable -> {
            Thread thread = new Thread(runnable);
            thread.setName("firebase-async-thread-" + Thread.currentThread().getId());
            thread.setDaemon(false); // 关键：非守护线程保证回调执行完成
            return thread;
        });
    }

    @Override
    public ExecutorService getExecutorService() {
        return fixedThreadPool;
    }

    // 应用关闭时释放线程池资源
    public void shutdownThreadPool() {
        fixedThreadPool.shutdown();
        try {
            if (!fixedThreadPool.awaitTermination(30, TimeUnit.SECONDS)) {
                fixedThreadPool.shutdownNow();
            }
        } catch (InterruptedException e) {
            fixedThreadPool.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}

2. 保证FirebaseApp全局唯一初始化 #

重复初始化FirebaseApp会覆盖之前的配置，导致自定义ThreadManager不生效。确保只初始化一次：

// 检查是否已初始化，避免重复创建
if (FirebaseApp.getApps().isEmpty()) {
    FirebaseOptions options = FirebaseOptions.builder()
            .setCredentials(GoogleCredentials.fromStream(new FileInputStream(new File(filePath))))
            .setThreadManager(new CustomThreadManager(10))
            .build();
    FirebaseApp.initializeApp(options);
}

// 获取绑定了自定义线程池的FirebaseMessaging实例
FirebaseMessaging messaging = FirebaseMessaging.getInstance();

3. 正确绑定回调到自定义线程池 #

sendEachForMulticastAsync返回ListenableFuture，需使用Firebase依赖的Guava工具绑定回调到自定义线程池，确保回调被执行：

ListenableFuture<List<SendResponse>> sendFuture = messaging.sendEachForMulticastAsync(multicastMessage);

// 添加回调并指定自定义线程池执行
Futures.addCallback(sendFuture, new FutureCallback<List<SendResponse>>() {
    @Override
    public void onSuccess(List<SendResponse> responses) {
        // 处理批量发送结果
        for (int i = 0; i < responses.size(); i++) {
            SendResponse response = responses.get(i);
            if (response.isSuccessful()) {
                // 第i条消息发送成功
            } else {
                // 第i条消息发送失败，获取错误信息
                Exception error = response.getException();
                // 处理错误逻辑
            }
        }
    }

    @Override
    public void onFailure(Throwable throwable) {
        // 处理全局发送失败（如参数错误、认证失败等）
        throwable.printStackTrace();
    }
}, messaging.getThreadManager().getExecutorService());

常见问题点 #

• 线程池使用守护线程：导致主线程退出后，回调线程被强制终止，无法执行完成。
• 未正确添加回调：忘记绑定FutureCallback，或使用外部线程池导致回调不生效。
• FirebaseApp重复初始化：后续初始化覆盖了自定义ThreadManager配置。

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