python-telegram-bot子线程运行触发signal报错问题咨询

我有一个进程需要接收两个不同来源的请求，请求内容可简化为普通字符串消息。两个来源的请求都会存入PriorityQueue，由主进程统一处理。其中一个请求源是基于python-telegram-bot包搭建的Telegram机器人，两个请求源都需要运行自身的事件循环才能产出请求，因此我计划将二者放在独立线程中启动。

对应设计思路的伪代码如下：

queue = PriorityQueue()
handler = RequestHandler(queue)
telegramRequester = TelegramRequester(queue)
anotherRequester = SomeOtherSourceRequester(queue)

telegramRequester.start()        # 启动Telegram机器人轮询/空闲循环
anotherRequester.start()         # 启动另一个请求源的事件循环
handler.handleRequestsLoop()     # 启动请求处理循环

Telegram机器人及对应请求器的实现大致如下：

class Bot:
    def __init__(self):
        self._updater = telegram.ext.Updater("my api token", use_context=True)
    
    def run(self):
        self._updater.start_polling(drop_pending_updates=True)
        self._updater.idle()

    def otherFunctions(self):
        # 包含命令注册、命令处理器绑定等逻辑
        # 机器人本身的功能已经测试通过，可正常运行


class TelegramRequester:
    def __init__(self, queue:RequestQueue) -> None:
        self._queue:RequestQueue = requestQueue
        self._bot = Bot()
        self._thread:threading.Thread = threading.Thread(target=self._bot.run)

    def start(self):
        if not self._thread.is_alive():
            self._thread.start()

运行上述代码时抛出如下错误：

File "...\myscript.py", line 83, in run
    self._updater.idle()
File "...\env\lib\site-packages\telegram\ext\updater.py", line 885, in idle
    signal(sig, self._signal_handler)
File "C:\Program Files\aaaProgrammieren\Python\Python3_9\lib\signal.py", line 47, in signal
    handler = _signal.signal(_enum_to_int(signalnum), _enum_to_int(handler))
ValueError: signal only works in main thread of the main interpreter

我是第一次接触Telegram API，也是第一次开发多线程并行程序，没有Web/网络编程相关经验。想确认是Telegram机器人本身就不支持在独立线程运行，还是我遗漏了某些简单配置，就可以实现上述多请求源的架构设计？

报错和Telegram机器人本身是否支持子线程运行无关，核心是Updater.idle()方法的默认机制限制：

• idle()默认会注册SIGINT、SIGTERM这类系统信号处理器，用来监听终止信号实现优雅退出
• Python的信号处理器只允许在主线程注册，子线程调用signal.signal()会直接抛出上述ValueError

不需要调整整体的多请求源架构，只需要修改Bot的run方法逻辑，避开在子线程调用信号注册逻辑即可，有两种常用实现方式：

方案1：手动实现子线程阻塞等待，替换idle()调用 #

idle()除了注册信号处理，核心作用就是阻塞线程保持运行，直到收到终止信号。在子线程运行时可以自行实现阻塞逻辑，不需要调用idle()：

import time
from threading import Event

class Bot:
    def __init__(self):
        self._updater = telegram.ext.Updater("my api token", use_context=True)
        self._stop_event = Event()
    
    def run(self):
        self._updater.start_polling(drop_pending_updates=True)
        # 循环阻塞等待停止信号，替换原idle()逻辑
        while not self._stop_event.is_set():
            time.sleep(0.5)
        # 收到停止信号后主动释放updater资源
        self._updater.stop()

    def stop(self):
        self._stop_event.set()

这种方式可控性强，可以在主线程需要退出时主动调用bot.stop()触发子线程清理逻辑，避免僵尸线程。

方案2：调用idle()时关闭信号注册参数 #

v13及以上版本的python-telegram-bot已经提供了原生参数支持关闭信号注册，只需要在调用idle()时传入signals=()即可：

def run(self):
    self._updater.start_polling(drop_pending_updates=True)
    # 传入空信号列表，不会尝试注册信号处理器
    self._updater.idle(signals=())

这种方式改造成本最低，不需要额外编写阻塞逻辑，注意自行处理程序退出时的机器人资源释放即可。

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