如何实现Python多进程间通信？Telegram Bot停止脚本场景

你的问题核心是多进程拥有独立的内存空间，普通的全局列表messages在子进程里会被复制一份，所以telegram_read里对列表的修改无法被check进程看到。要实现进程间通信，Python的multiprocessing模块提供了多种工具，下面给你两种简单可行的方案：

方案1：使用Manager.list()创建共享列表 #

Manager可以创建跨进程共享的对象，它会在主进程启动一个服务进程，其他子进程通过代理访问这个共享对象。

修改后的代码：

from multiprocessing import Process, Manager
from telethon import TelegramClient, events
import time

def stop_script():
    # 这里实现你停止脚本的逻辑，比如终止进程、清理资源等
    print("脚本已停止")

def telegram_read(messages):
    client = TelegramClient('anon', API_ID, API_HASH)

    @client.on(events.NewMessage(chats=BOT_NAME))
    async def my_event_handler(event):
        messages.append(event.raw_text)
        print(f"收到消息：{event.raw_text}")

    client.start()
    client.run_until_disconnected()

def check(messages):
    while True:
        if 'STOP' in messages:
            stop_script()
            return
        time.sleep(10)

if __name__ == "__main__":
    with Manager() as manager:
        # 创建跨进程共享的列表
        messages = manager.list()
        p1 = Process(target=telegram_read, args=(messages,))
        p1.start()
        p2 = Process(target=check, args=(messages,))
        p2.start()
        p1.join()
        p2.join()

关键修改点： #

• 导入Manager类，用manager.list()替代普通列表
• 将共享列表作为参数传递给两个子进程（不能直接用全局变量，因为子进程无法共享主进程的全局变量）
• 补充实现stop_script函数（原代码未定义，需填充你的停止逻辑）

方案2：使用Queue实现消息传递 #

如果只是用来传递停止指令，Queue（队列）更适合，它是进程安全的，专门用于进程间的消息传递。

修改后的代码：

from multiprocessing import Process, Queue
from telethon import TelegramClient, events
import time

def stop_script():
    print("脚本已停止")

def telegram_read(msg_queue):
    client = TelegramClient('anon', API_ID, API_HASH)

    @client.on(events.NewMessage(chats=BOT_NAME))
    async def my_event_handler(event):
        msg_queue.put(event.raw_text)
        print(f"收到消息：{event.raw_text}")

    client.start()
    client.run_until_disconnected()

def check(msg_queue):
    while True:
        # 非阻塞式检查队列，避免卡住
        while not msg_queue.empty():
            msg = msg_queue.get()
            if msg == 'STOP':
                stop_script()
                return
        time.sleep(10)

if __name__ == "__main__":
    msg_queue = Queue()
    p1 = Process(target=telegram_read, args=(msg_queue,))
    p1.start()
    p2 = Process(target=check, args=(msg_queue,))
    p2.start()
    p1.join()
    p2.join()

关键修改点： #

• 导入Queue类，创建队列对象传递给子进程
• telegram_read把收到的消息放入队列，check进程从队列中取出消息检查
• 使用非阻塞方式遍历队列，避免因为队列空而阻塞

注意事项 #

• 子进程无法直接共享主进程的全局变量，必须通过Manager、Queue、Pipe等进程间通信工具传递数据
• 所有共享对象必须是可序列化的（pickleable），这里我们只传递文本消息，不存在问题
• 如果需要更优雅的停止逻辑，可以使用multiprocessing.Event来发送停止信号，避免轮询检查

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