RxPython 4.2：如何将后台线程数据切换至主线程处理？

问题根源：CurrentThreadScheduler的工作逻辑 #

你用CurrentThreadScheduler没达到预期，是因为它的调度规则不是"固定绑定创建时的线程"，而是在调用任务调度的线程上执行任务。

你的代码里，my_subject.on_next(data)是在后台线程（Dummy-7）里触发的，observe_on(CurrentThreadScheduler())会把订阅回调任务放到当前触发线程的执行队列里，自然就跟着后台线程跑了，这就是为什么订阅和回调的线程一致。

解决方案：让主线程拥有可处理任务的调度循环 #

要把订阅回调调度到主线程，必须让主线程运行一个事件循环来接收并处理调度任务。RxPy里有两种常用方式：

方式1：使用BlockingScheduler（推荐）

BlockingScheduler会在当前线程（主线程）启动一个阻塞式事件循环，自动处理调度队列里的任务：

from reactivex import operators as ops
from reactivex.scheduler import BlockingScheduler
from reactivex import Subject
import threading
from threading import Thread

class MyThread(Thread):
    def __init__(self, callback):
        Thread.__init__(self)
        self.callback = callback

    def run(self):
        self.callback("hello")

my_subject = Subject()

def callback(data):
    print(f"In callback: {threading.current_thread().name}")
    my_subject.on_next(data)

if __name__ == "__main__":
    # 初始化主线程调度器
    main_thread_scheduler = BlockingScheduler()

    print(f"Before stream: {threading.current_thread().name}")

    background_thread = MyThread(callback=callback)
    
    # 指定用主线程调度器处理订阅回调
    my_subject.pipe(
        ops.observe_on(main_thread_scheduler)
    ).subscribe(
       lambda x: print(f"In subscription: {threading.current_thread().name}")
    )
    
    background_thread.start()
    
    # 启动主线程的事件循环，阻塞处理任务
    main_thread_scheduler.start()

执行后输出符合预期：

Before stream: MainThread
In callback: Dummy-7
In subscription: MainThread

方式2：手动触发调度队列刷新

如果不想用阻塞式循环，可以手动在主线程循环中调用flush()处理任务队列：

from reactivex import operators as ops
from reactivex.scheduler import Scheduler
from reactivex import Subject
import threading
from threading import Thread
import time

class MyThread(Thread):
    def __init__(self, callback):
        Thread.__init__(self)
        self.callback = callback

    def run(self):
        self.callback("hello")

my_subject = Subject()

def callback(data):
    print(f"In callback: {threading.current_thread().name}")
    my_subject.on_next(data)

if __name__ == "__main__":
    main_thread_scheduler = Scheduler()

    print(f"Before stream: {threading.current_thread().name}")

    background_thread = MyThread(callback=callback)
    
    my_subject.pipe(
        ops.observe_on(main_thread_scheduler)
    ).subscribe(
       lambda x: print(f"In subscription: {threading.current_thread().name}")
    )
    
    background_thread.start()
    
    # 主线程主动循环处理调度任务
    while True:
        main_thread_scheduler.flush()
        time.sleep(0.1)

关键总结 #

• CurrentThreadScheduler的"当前线程"指的是触发任务调度的线程，不是创建调度器的线程；
• 要让任务回到主线程，必须让主线程有一个持续运行的事件循环来处理调度队列，BlockingScheduler是最简便的实现方式。

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