Python如何从指定URL持续拉取数据流并实时写入本地文件

问题原因 #

原有代码卡住的原因是requests默认会等待完整响应体全部加载完成后才会返回请求结果，对于永不中断的持续数据流，自然会一直阻塞不会终止。

正确实现方案 #

你需要开启requests的流式请求模式，拿到响应后按块迭代读取实时到达的内容，同步写入文件即可。

基础示例代码 #

import requests

# 注意补全URL的http/https协议头，否则会请求报错
STREAM_URL = "http://www.someurl.com/api/getdata?password=..."

with requests.Session() as s:
    # 核心参数stream=True：仅加载响应头就返回，不等待完整响应体
    resp = s.get(STREAM_URL, stream=True)
    # 校验请求状态正常
    resp.raise_for_status()
    # 以追加模式打开文件，也可根据需求用w覆盖模式
    with open("stream_output.txt", "a", encoding="utf-8") as f:
        # 按行迭代响应内容，适配文本类型的持续流
        for line in resp.iter_lines(decode_unicode=True):
            if line:  # 过滤空行
                f.write(line + "\n")
                # 立即刷新缓冲区，确保内容实时写入磁盘
                f.flush()

固定字节块读取写法 #

如果不需要按行处理，想要按固定字节大小读取，可以使用iter_content方法：

# 每次读取1024字节（1KB）内容
for chunk in resp.iter_content(chunk_size=1024, decode_unicode=True):
    if chunk:
        f.write(chunk)
        f.flush()

实用优化说明 #

• 如果服务端返回非UTF-8编码内容，需要在写入文件时指定对应编码，避免出现乱码
• 可以根据存储和性能需求调整chunk_size大小，块越小写入越实时，对应的IO开销也越高
• 可以增加异常捕获和重试逻辑，适配网络中断、手动停止等场景，降低数据丢失风险：

import requests
import time

STREAM_URL = "http://www.someurl.com/api/getdata?password=..."
retry_interval = 5  # 连接断开后5秒重试

while True:
    try:
        with requests.Session() as s:
            # 连接超时设为5秒，读取超时设为None适配长连接场景
            resp = s.get(STREAM_URL, stream=True, timeout=(5, None))
            resp.raise_for_status()
            with open("stream_output.txt", "a", encoding="utf-8") as f:
                for line in resp.iter_lines(decode_unicode=True):
                    if line:
                        f.write(line + "\n")
                        f.flush()
    except (requests.exceptions.RequestException, ConnectionError) as e:
        print(f"连接异常，{retry_interval}秒后重试：{e}")
        time.sleep(retry_interval)
    except KeyboardInterrupt:
        print("手动停止任务")
        break

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