SSE传输数据

1、yield讲解 

1. yield 和 return 的区别

（1）yield

• yield 用于定义生成器函数。

• 生成器函数会返回一个生成器对象，每次调用 next() 时，函数会执行到 yield 语句并返回 yield 后面的值，然后暂停执行，直到下一次调用 next()。

• 生成器函数适合用于逐步生成数据（如 SSE 的实时数据传输）。

（2）return

• return 用于从函数中返回值并结束函数的执行。

• 一旦函数执行到 return，函数会立即结束，后续代码不会执行。

---

2. 为什么不能同时使用 yield 和 return

• 生成器函数的特性：

  • 生成器函数只能使用 yield 返回值，不能使用 return 返回值。

  • 如果生成器函数中使用 return，它只能用于结束函数执行，而不能返回值（Python 3.3+ 中可以使用 return value，但返回值会被忽略）。

• 普通函数的特性：

  • 普通函数可以使用 return 返回值，但不能使用 yield。

---

3. 实现 SSE 传输的正确方式

SSE 传输需要逐步生成数据并发送到客户端，因此必须使用 yield，而不能使用 return。

import subprocessdef generate_logs():process = subprocess.Popen(["ping", "google.com"],stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True,bufsize=1,universal_newlines=True)while True:output = process.stdout.readline()if output == "" and process.poll() is not None:break  # 如果没有输出且进程已结束，退出循环if output:yield f"data: {output}\n\n"  # 逐步发送数据error = process.stderr.readline()if error:yield f"data: {error}\n\n"  # 逐步发送错误yield "data: Process completed.\n\n"  # 进程结束后发送完成信号

1. SSE 数据传输格式：

   • SSE 要求数据以 data: <内容>\n\n 的格式发送。

   • 如果需要传输对象，必须将对象序列化为字符串（如 JSON 格式）。

（1）yield 的特性

• yield 是 Python 中生成器的关键字，用于逐步生成数据。

• 生成器是单线程的，只能在主线程中运行。

• 如果在子线程中使用 yield，会导致生成器无法正常工作，因为生成器的状态无法跨线程共享。

（2）多线程的限制

• Python 的全局解释器锁（GIL）限制了多线程的并行执行。

• 生成器的状态（如局部变量、执行位置）是线程私有的，无法在多个线程之间共享。

（3）SSE 的实现

• SSE 是基于 HTTP 的长连接，要求服务器通过 yield 逐步生成数据并发送到客户端。

• 如果 yield 被放入子线程中，主线程无法捕获生成器的返回值，导致 HTTP 响应无法正常发送。

1. SSE 协议的分帧机制：

   • SSE 协议要求每条消息以 \n\n 结尾。

   • 如果发送的数据中包含 \n，SSE 客户端（如 fetchEventSource）可能会将单次发送的数据拆分为多个消息。

2. 后端发送的数据包含换行符：

   • 如果 buffer 中的内容包含 \n，SSE 客户端会将其视为消息分隔符，从而触发多次 onmessage。

3. 网络传输分块：

   • 如果数据量较大，网络层可能会将数据拆分为多个 TCP 包，导致客户端多次接收。

 2、subprocess讲解

 

1. subprocess.Popen() 的基本用法

subprocess.Popen() 用于启动子进程，并可以通过参数 stdout 和 stderr 分别捕获标准输出和标准错误。

常用参数：

• stdout：指定标准输出的处理方式。

  • subprocess.PIPE：捕获输出。

  • subprocess.DEVNULL：丢弃输出。

  • 文件对象：将输出写入文件。

• stderr：指定标准错误的处理方式（同上）。

• text=True：以文本模式读取输出（Python 3.7+）。

• bufsize=1：行缓冲，逐行读取输出。

---

2. 区分 stdout 和 stderr

通过 process.stdout 和 process.stderr 可以分别读取标准输出和标准错误。

示例代码：

import subprocess# 启动子进程
process = subprocess.Popen(["ls", "/nonexistent"],  # 命令（列出不存在的目录）stdout=subprocess.PIPE,  # 捕获标准输出stderr=subprocess.PIPE,  # 捕获标准错误text=True,              # 以文本模式读取输出bufsize=1               # 行缓冲
)# 读取标准输出
for line in process.stdout:print("标准输出:", line.strip())# 读取标准错误
for line in process.stderr:print("标准错误:", line.strip())# 等待进程结束
process.wait()

3. 逐行读取 stdout 和 stderr

如果需要同时读取 stdout 和 stderr，可以使用 process.stdout.readline() 和 process.stderr.readline()。

示例代码：

import subprocess# 启动子进程
process = subprocess.Popen(["ping", "google.com"],  # 命令stdout=subprocess.PIPE,  # 捕获标准输出stderr=subprocess.PIPE,  # 捕获标准错误text=True,              # 以文本模式读取输出bufsize=1               # 行缓冲
)# 逐行读取输出
while True:# 读取标准输出output = process.stdout.readline()if output == "" and process.poll() is not None:break  # 如果没有输出且进程已结束，退出循环if output:print("标准输出:", output.strip())# 读取标准错误error = process.stderr.readline()if error:print("标准错误:", error.strip())# 等待进程结束
process.wait()

4. 使用 communicate() 一次性读取

process.communicate() 可以一次性读取所有标准输出和标准错误。

示例代码：

import subprocess# 启动子进程
process = subprocess.Popen(["ls", "/nonexistent"],  # 命令stdout=subprocess.PIPE,  # 捕获标准输出stderr=subprocess.PIPE,  # 捕获标准错误text=True               # 以文本模式读取输出
)# 一次性读取所有输出
stdout, stderr = process.communicate()print("标准输出:", stdout)
print("标准错误:", stderr)

5. 注意事项

1. 缓冲区问题：

   • 如果子进程的输出量较大，可能会导致缓冲区阻塞。可以通过设置 bufsize=1 或使用 process.stdout.readline() 逐行读取。

2. 死锁风险：

   • 如果同时读取 stdout 和 stderr，可能会导致死锁。可以使用 process.communicate() 避免死锁。

3. 编码问题：

   • 如果输出包含非 ASCII 字符，确保设置 text=True 或指定正确的编码（如 encoding="utf-8"）。