如何基于Python requests与concurrent.futures实现高吞吐实时流的无丢失定时处理

原来的串行实现把三个完全独立的逻辑放在了同一个执行循环里：

• 持续从requests流迭代读取标识字符串
• 按n秒窗口做计数统计
• 同步执行耗时的foo算法
  方案存在两个无法修复的硬伤：

1. 流迭代循环被foo阻塞后，TCP接收缓冲区会被未读取的流数据填满，最终触发服务端暂停推送甚至断连，直接丢失流数据
2. 当foo执行时长超过n秒时，循环暂停的时段内所有产生的流数据都无法被统计，完全无法满足高吞吐场景的无丢失要求

核心思路是做完全的执行链路隔离，绝对不能让任何非流读取的操作阻塞iter_lines的迭代过程，可结合concurrent.futures、线程安全队列实现解耦，不需要引入重型流处理框架就能满足需求。

推荐架构逻辑 #

• 独立的流读取线程：唯一负责持续迭代读取request.iter_lines()的内容，只做最轻量的窗口计数操作，不执行任何耗时任务
• 线程安全的解耦缓冲区：存储每个时间窗口统计完成的计数快照，实现流读取逻辑和计算逻辑的解耦
• 独立的计算工作线程：从缓冲区取窗口统计结果，执行foo算法，和流读取线程完全并行

关键实现规则 #

• 窗口计数操作的开销要做到极致轻量：每次迭代拿到字符串后，只做字典计数累加，到了n秒窗口时间点，就把当前的计数字典拷贝成独立快照，丢进缓冲区后立刻重置计数器开始下一个窗口的统计，全程不等待缓冲区的消费进度
• 流读取线程全程禁止阻塞：不要在流迭代的循环里加锁、做磁盘/网络IO、等待计算任务完成，从根源上保证iter_lines的迭代循环始终能及时读取缓冲区里的流数据，避免TCP层堆积丢数
• 兼容foo的超时/积压场景：就算foo单次执行时长超过n秒，因为每个窗口的结果都是独立存入缓冲区的，计算线程会按顺序消费，既不会打断流读取，也不会丢失窗口统计结果；如果缓冲区积压长度超过阈值，可单独加告警逻辑，不反向影响流读取流程

具体实现参考 #

可直接使用concurrent.futures.ThreadPoolExecutor做简单解耦，不需要复杂的队列实现也能满足需求：

from concurrent.futures import ThreadPoolExecutor
import time
import requests

# 初始化独立线程池，专门跑耗时的foo任务
executor = ThreadPoolExecutor(max_workers=2)

request = requests.get(URL, stream=True)
# 开启unicode解码，直接拿到字符串，避免额外的解码开销
stream = request.iter_lines(decode_unicode=True)

def stream_reader_loop(window_seconds):
    while True:
        window_start = time.time()
        counter = {}
        # 流读取+窗口统计的循环，全程不调用foo
        for line in stream:
            counter[line] = counter.get(line, 0) + 1
            # 到点就结束当前窗口统计
            if time.time() - window_start >= window_seconds:
                break
        # 把foo任务提交到独立线程池，提交完立刻回到流读取循环，不等待foo执行完成
        executor.submit(foo, counter)

if __name__ == "__main__":
    # 按实际窗口大小传入
    stream_reader_loop(window_seconds=n)

高吞吐场景注意事项 #

• 不要使用多进程做流读取解耦，requests的流迭代器不是可序列化的对象，跨进程传递会直接断连丢数据，多线程是最优选择
• 不要拆分出独立的定时器线程做窗口切分，那样需要跨线程传递计数器，引入锁竞争的开销，高吞吐下会明显降低读取效率；直接在流迭代循环里每次判断时间戳切分窗口，逻辑简单且开销可以忽略
• 所有foo执行过程中的异常捕获逻辑都要放在foo内部，不要让线程池的异常反向传递到流读取线程，避免流读取崩溃断连

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