如何将GCP Pub/Sub StreamingPullFutures与discordpy异步集成？

问题背景 #

需要用GCP Pub/Sub的StreamingPullFutures订阅功能配合discordpy，实现接收外部指令（如移除用户、跨服务器发送更新），并希望在discord机器人启动的on_ready事件中启动监听逻辑。尝试两种方式均遇问题：

1. 单独编写pub_sub_function但无法正常监听消息；
2. 直接在on_ready中调用subscriber_client.subscribe(...).result()，导致机器人事件循环阻塞，触发心跳超时警告：

WARNING  discord.gateway Shard ID None heartbeat blocked for more than 10 seconds.
Loop thread traceback (most recent call last):

解决方案 #

核心思路是将Pub/Sub的监听逻辑放到独立线程中运行，避免阻塞discordpy的异步事件循环；同时在Pub/Sub回调中调用discordpy异步方法时，需通过机器人的事件循环调度，保证线程安全。

完整代码示例 #

import threading
import json
from google.cloud import pubsub_v1
import discord
from discord.ext import commands

# 初始化discord机器人，根据需求开启对应意图
intents = discord.Intents.default()
intents.members = True  # 处理用户移除需开启成员意图
bot = commands.Bot(command_prefix='!', intents=intents)

def run_pubsub_listener():
    """在独立线程中运行Pub/Sub订阅监听"""
    subscriber_client = pubsub_v1.SubscriberClient()
    # 替换为你的项目ID和订阅名称
    subscription = subscriber_client.subscription_path('my-project-id', 'my-subscription')

    def callback(message):
        """处理Pub/Sub消息的回调函数"""
        print(f"收到Pub/Sub消息: {message}")
        try:
            # 解析消息内容（假设消息为JSON格式，可根据实际调整）
            message_data = json.loads(message.data.decode('utf-8'))
            instruction = message_data.get("instruction")
            target_guild_id = message_data.get("guild_id")
            target_channel_id = message_data.get("channel_id")
            target_user_id = message_data.get("user_id")

            # 定义异步处理discord操作的函数
            async def handle_discord_task():
                guild = bot.get_guild(target_guild_id)
                if not guild:
                    print(f"未找到服务器ID: {target_guild_id}")
                    return

                # 根据指令类型执行对应操作
                if instruction == "send_update":
                    channel = guild.get_channel(target_channel_id)
                    if channel:
                        await channel.send(message_data.get("content", "收到更新通知"))
                elif instruction == "remove_user":
                    member = guild.get_member(target_user_id)
                    if member:
                        await member.kick(reason=message_data.get("reason", "外部指令移除"))

            # 将异步任务提交到机器人的事件循环中执行
            bot.loop.call_soon_threadsafe(lambda: bot.loop.create_task(handle_discord_task()))
            
        except Exception as e:
            print(f"处理消息出错: {e}")
        finally:
            message.ack()  # 确认消息已处理

    future = subscriber_client.subscribe(subscription, callback)
    try:
        future.result()  # 线程阻塞直到订阅取消
    except KeyboardInterrupt:
        future.cancel()
        future.result()

@bot.event
async def on_ready():
    print(f'{bot.user} ({bot.user.id}) 已启动！')
    # 启动Pub/Sub监听线程，设置为守护线程随主程序退出
    pubsub_thread = threading.Thread(target=run_pubsub_listener, daemon=True)
    pubsub_thread.start()

# 替换为你的机器人令牌
bot.run('YOUR_BOT_TOKEN')

关键说明 #

1. 线程隔离：用threading.Thread启动Pub/Sub监听，避免阻塞discordpy的异步事件循环，解决心跳超时问题；设置daemon=True让线程随主程序自动终止。
2. 异步任务调度：Pub/Sub回调运行在非事件循环线程中，不能直接调用discordpy的异步方法，需通过bot.loop.call_soon_threadsafe将异步任务提交到机器人的事件循环执行，保证线程安全。
3. 消息解析：示例假设消息为JSON格式，实际可根据业务需求调整解析逻辑。
4. 权限与意图：确保机器人拥有对应操作的权限（如踢人、发送消息），并在Discord开发者后台开启所需的意图（如成员意图）。

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