使用GMAIL API Pub/Sub时回调触发新建线程的问题咨询

看起来你遇到了Pub/Sub客户端默认配置引发的典型问题：每次新消息触发回调时，客户端会无限制创建新线程，短时间内线程数量爆炸，最终把CPU资源吃光。而且你已经排除了业务逻辑（mp.get_messages_of_interest()）的影响，那问题肯定出在Pub/Sub的线程管理上。

问题根源 #

Google Pub/Sub的Python客户端默认使用无界线程池来处理回调——只要有新消息进来，就会新建线程去执行回调函数，哪怕之前的线程还在运行，或者线程池里已经堆了一堆空闲线程。当Gmail的Webhook触发频繁时，线程数会快速飙升，CPU被大量线程的上下文切换和调度耗尽。

解决方案 #

下面是几个针对性的修复步骤，按优先级排序：

1. 给Pub/Sub客户端设置有界线程池

这是最直接的解决办法，通过限制线程池的最大线程数，从根源上控制并发线程数量。

修改你的代码如下：

from concurrent.futures import ThreadPoolExecutor
import google.cloud.pubsub_v1 as pubsub

# 自定义线程池，比如最多允许10个并发线程（根据你的服务器配置调整）
executor = ThreadPoolExecutor(max_workers=10)

# 创建SubscriberClient时传入自定义线程池
subscriber = pubsub.SubscriberClient(executor=executor)
topic = "myTopic"
subscription_name = "mySub"
subscription = subscriber.subscribe(subscription_name)

def callback(message):
    print("Let's GO !!")
    # mp.get_messages_of_interest()  # 即使注释掉，线程数也会被合理控制
    message.ack()

future = subscription.open(callback)
try:
    # 阻塞主线程，同时捕获中断信号以便优雅退出
    future.result()
except KeyboardInterrupt:
    future.cancel()

2. 确保消息确认逻辑可靠

虽然你已经调用了message.ack()，但如果回调过程中抛出异常，ack()可能不会被执行，导致Pub/Sub认为消息处理失败，会重新推送这条消息——这会进一步增加线程创建的频率。

给回调加个异常捕获，确保ack()或nack()总能被调用：

def callback(message):
    try:
        print("Let's GO !!")
        # mp.get_messages_of_interest()
        message.ack()
    except Exception as e:
        print(f"处理消息出错: {str(e)}")
        # 处理失败时，用nack让Pub/Sub稍后重新推送（根据业务需求调整）
        message.nack()

3. 限制Pub/Sub订阅的并发推送量

除了客户端线程池，你还可以在Pub/Sub订阅层面设置并发限制，避免一次性收到太多消息。比如限制同时未确认的消息数量：

如果是创建新订阅，可以这样设置：

subscriber.create_subscription(
    name=subscription_name,
    topic=topic,
    max_outstanding_messages=10,  # 最多同时处理10条未确认消息
    max_outstanding_bytes=10*1024*1024  # 限制未确认消息的总大小为10MB
)

如果是已有订阅，也可以通过Google Cloud Console找到对应订阅，修改这两个参数。

总结 #

通过上述三个步骤，你就能把线程数量控制在合理范围内，解决CPU消耗过高的问题。核心思路就是从客户端线程池、消息确认可靠性、服务端推送限制三个维度，把并发量压到服务器能承受的水平。

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