如何在Google Pub/Sub中快速发布多条消息（无需多进程/多线程）

你当前性能过低的核心原因是每次调用publisher.publish()后立即执行future.result()，同步阻塞等待单条消息的发布结果，完全浪费了客户端的批量发布能力，每条消息都单独发起一次网络请求，所以速度只能维持在每秒几十条的水平。

以下是无需多进程/多线程的优化方案：

• 取消单条消息的同步等待，先收集所有发布任务的future，全部提交后再统一等待结果：这是收益最高的优化点，调整后客户端可以正常按照你配置的BatchSettings攒批发送，一批最多1000条消息只需要走一次网络请求，吞吐量会大幅提升。
• 调整批量配置参数：你当前设置的max_latency=0.1（100ms）比默认值10ms更长，相当于客户端会等待更长时间攒批。如果你的消息生成速度足够快，可以适当调低这个值降低延迟；如果消息量不大但优先保障吞吐量，可以保持当前配置甚至适当调高。
• 开启传输压缩：在PublisherClient配置中开启gzip压缩，你发送的是JSON格式数据，压缩率通常可以达到70%以上，大幅减少网络传输耗时。
• 优化序列化效率：把标准库的json替换为性能更高的ujson库，序列化速度可以提升2~5倍；也可以提前批量完成所有数据的序列化，避免在发布循环中做序列化操作占用时间。
• 调整客户端流控配置：通过publisher_options参数调整客户端待发布消息队列的阈值，避免队列满导致发布阻塞，可根据实际业务消息量级调大待发布队列长度。

优化后的代码示例：

import ujson  # 替换标准json库提升序列化速度
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.types import BatchSettings, PublisherOptions, FlowControl

# 初始化客户端时添加压缩和流控配置
publisher = pubsub_v1.PublisherClient(
    credentials=credentials,
    batch_settings=BatchSettings(
        max_messages=1000,
        max_bytes=1 * 1000 * 1000,
        max_latency=0.01,  # 可根据实际场景调整为合适的攒批等待时长
    ),
    publisher_options=PublisherOptions(
        flow_control=FlowControl(
            max_messages=10000,  # 调大待发布队列长度，适配大流量场景
        )
    ),
    client_options={
        "grpc.enable_gzip": True  # 开启gzip压缩，减少传输数据量
    }
)

topic_name = f'projects/{PROJECT_ID}/topics/{TOPIC_PUBSUB}'

futures = []
# 先提交所有发布任务，不等待单条结果，让客户端可以正常攒批
for data in results:
    bytes_json_data = ujson.dumps(data).encode()
    future = publisher.publish(topic_name, bytes_json_data)
    futures.append(future)

# 全部提交完成后，统一等待所有任务执行完成
for future in futures:
    future.result()

按照上述方案优化后，正常场景下发布吞吐量可以提升到每秒数千条甚至更高，完全适配单线程运行的环境要求。

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