03-采集线程与发送协程Outbox邮箱与背压设计前面两篇讲控制端Viewer这篇把镜头转向被控端Agent。被控端要把屏幕采出来、编码、加密、发走这条链路上有个经典矛盾采集是 CPU 密集且阻塞的活不能放到 asyncio 事件循环里可最后发网络又必须在事件循环里做。怎么把「阻塞的采集线程」和「异步的发送协程」严丝合缝地接起来答案就是项目源码里的Outbox邮箱以及一整套背压backpressure设计。一、完整的线程模型图先上全貌被控端运行起来是这样的采集/编码线程阻塞 事件循环asyncio ───────────────────── ────────────────────── grab → diff → atlas → JPEG → Outbox → 发送协程 → WebSocket ↑ 接收协程处理 CTRL / INPUT watchdog自上而下看采集/编码线程独立线程死循环里不停地grab抓屏、diff差分、atlas拼图、JPEG编码把成品丢进Outbox。Outbox线程安全的「邮箱」采集线程往里塞发送协程从里头取。发送协程跑在 asyncio 事件循环里从 Outbox 取包加密后通过 WebSocket 发出去。接收协程 watchdog同样在事件循环里负责处理远端控制消息请求关键帧、键鼠输入等和急停/空闲保命逻辑。关键分工是重活采集编码在普通线程网络收发在 asyncio。两者通过 Outbox 这个缓冲区解耦谁也不阻塞谁。二、为什么采集与编码必须放独立线程屏幕采集和差分编码有两大特征CPU 密集实测原生分辨率下分块差分就要十几毫秒整条采集编码链路接近一帧 30ms。如果放在 asyncio 事件循环里同步执行循环会被长时间占住期间什么都干不了——收不到远端消息、发不出已排队的键鼠事件、连保活 ping 都发不出去。阻塞且不可让出像grab()这种调用底层抓取接口是阻塞式的不像await那样能中途让出。asyncio 的协程一旦进了这种函数就卡死整个循环。所以项目源码把采集编码包成一个threading.Thread名字叫capture守护线程它自己转自己的循环和事件循环互不干扰。唯一的接口就是那个 Outbox。三、Outbox一个线程安全的发送邮箱Outbox的代码不长但每句话都有讲究。项目源码定义如下classOutbox:def__init__(self,maxlen,loop):self._itemscollections.deque()self._lockthreading.Lock()self._maxlenmax(1,maxlen)self._looploop self._eventasyncio.Event()self.dropped0self.sent0四个核心成员各有使命成员类型作用_itemsdeque存放待发数据包的队列_lockthreading.Lock保护队列的线程锁采集线程和发送协程都碰它_eventasyncio.Event唤醒发送协程「有新包了」_loop事件循环用来跨线程安全地唤醒「线程安全」不是靠运气是靠这把锁put采集线程调用和get发送协程调用都对_items加锁操作两个线程不会同时改队列导致数据错乱。四、跨线程唤醒call_soon_threadsafe采集线程往 Outbox 塞完包得告诉发送协程「有活干了」。但这个「告诉」不能简单粗暴地直接调用协程函数——你在另一个线程里直接碰事件循环的对象是线程不安全的。项目源码用loop.call_soon_threadsafe(event.set)来唤醒defput(self,packet,tiles):stale[]withself._lock:iflen(self._items)self._maxlen:_old_packet,old_tilesself._items.popleft()stalelist(old_tiles)self.dropped1self._items.append((packet,tiles))self._loop.call_soon_threadsafe(self._event.set)# 跨线程安全唤醒returnstalecall_soon_threadsafe是 asyncio 专门给「非事件循环线程」准备的接口它会把event.set()这个动作排进事件循环的下一次迭代里执行且内部自带线程同步绝不会和循环本身的调度打架。发送协程那边则是asyncdefget(self):whileTrue:withself._lock:ifself._items:returnself._items.popleft()self._event.clear()awaitself._event.wait()取不到就clear事件、然后await等它被唤醒。一唤醒就再检查队列。这套「锁 事件」的组合就是经典的线程安全生产者-消费者模型。五、容量满时丢最旧的包——保留最新画面状态Outbox 有容量上限配置项send_queue_max默认 8。网络一旦慢了、发送协程追不上采集线程的速度队列就会堆满。这时候怎么办一个反直觉但正确的决定容量满时丢掉最旧的包保留最新的画面状态。iflen(self._items)self._maxlen:_old_packet,old_tilesself._items.popleft()stalelist(old_tiles)self.dropped1self._items.append((packet,tiles))为什么要丢旧的而不是新的因为远程桌面是「实时画面」旧的那帧早就过时了你发过去用户也看的是旧世界而最新的帧才代表「现在」。丢了旧帧最多让画面短暂跳一下但如果为了不丢而积压旧帧延迟会越堆越高用户感觉到的就是「操作越来越卡、越来越滞后」那比偶尔跳帧糟多了。而且这里还埋了一个「丢帧一致性」的钩子——put返回被丢掉的那个包涉及的块stale采集线程拿到后会把这些块并回「待补发集合」下一帧重新发送。这样画面能自愈又不需要专门发整屏关键帧拥塞时发大帧只会雪上加霜。这是项目源码里一个很漂亮的设计细节宁可掉帧也不让画面永久残缺。六、LOOP_SLEEP_IDLE 0.002静止时避免空转烧 CPU采集线程的循环里有个细节值得单独说。当画面静止、grab()返回None没变化时线程不能「什么都不做地空转」——那会 100% 占满一个 CPU 核心风扇狂转。项目源码定义了LOOP_SLEEP_IDLE 0.002也就是 2 毫秒rgbcap.grab(forcewant_keyframe)ifrgbisNone:# 画面没有变化。静止时几乎零开销time.sleep(sleep_idle)continue画面静止时每次循环grab返回空线程就sleep(0.002)再继续。2 毫秒的间隔意味着每秒最多轮询 500 次既不会漏掉突然的变化dxcam 有变化时grab立刻返回又不会把 CPU 烧穿。这是个很务实的折中太频繁烧 CPU太稀疏又响应迟钝2ms 刚好。七、采集线程任何异常都不能悄悄死掉采集线程是「画面来源」它一旦崩了整个远程桌面就黑屏且毫无提示。所以项目源码对采集循环做了最外层兜底try:...# 差分、编码、投递 OutboxexceptExceptionase:# 采集线程里任何异常都不能让它悄悄死掉self.log([采集] 处理帧出错%s: %s%(type(e).__name__,e))注意这里 catch 的是「处理单帧」的异常循环本身不会被打断。哪怕某一帧的差分算出问题也只是丢掉这一帧并打日志下一帧继续走。采集线程绝不能因为一两个坏帧就退出——「不能让它悄悄死掉」是项目源码注释里原话也是长连接稳定性的硬要求。八、重连退避 1→2→4→30 秒封顶网络是会抖动的。被控端和远端的连接可能因为公司网络瞬断、VPS 重启而断开。项目源码的重连逻辑用指数退避避免「一断开就疯狂重连」把链路和自己都冲垮backoff1.0whilenotself.stop_flag.is_set():try:awaitself._run_once()backoff1.0# 连上了就重置except...:...# 退避等待awaitasyncio.sleep(backoff)backoffmin(30.0,backoff*2)退避序列是1 → 2 → 4 → 8 → 16 → 30到 30 秒就封顶不再涨。第一次断立刻 1 秒后重连响应快如果还连不上就翻倍拉长间隔减轻压力到 30 秒封顶不至于等太久。连上后backoff重置回 1.0下次从短间隔重新开始。这套退避对「公司网络偶发抖动」特别友好。九、断线必须松开按键release_all这条是保命相关的硬规则。设想一种情况你正远程按着某个键比如 Shift突然网络断了。如果这时候不把「按键抬起」发给远端远端那台电脑就会一直以为 Shift 还按着——你人已经断线了可公司电脑上的 Shift 卡死了后面的操作全乱套。所以项目源码在任何断线路径上都调release_all()把所有可能卡住的按键全部补发「抬起」# 断线时必须松开所有按键否则远端会留下「卡住的按键」self._teardown_injection()release_all()补发三个鼠标键的抬起外加一组常见修饰键扫描码的抬起覆盖左右 Ctrl、左右 Shift 等。无论正常断开、空闲超时断开、还是急停断开都走这一道。这是「宁可多抬一次不能卡一个键」的防御。十、只读判定注入器没建起来就一律当只读最后说一个判定逻辑它直接关系到安全。readonly只读不是看配置里写了什么而是看注入器到底有没有真正建起来readonlyself._injectorisNone# 注入器没建起来就一律当只读为什么这么判定因为「能不能注入键鼠」这件事最终取决于注入器对象是否成功创建。如果配置写了允许输入但注入器因为某些原因比如权限不够、依赖缺失没建起来那就一律按只读处理——绝不因为「配置说能注入」就硬着头皮往可能没初始化的注入器里塞事件。这个判定的好处是把「安全」建立在「实际能力」而非「纸面配置」上。而且这个只读状态会通过握手透传给控制端控制端据此在界面上用红/绿明确提示键鼠注入是否启用做到「所见即所得」。到这里被控端的「采集 → 编码 → 邮箱 → 发送」链路就讲透了独立线程扛重活Outbox 做线程安全缓冲满了丢旧的不丢新的静止时小睡避免烧 CPU异常兜住不悄死断线必松键只读看实际能力。