FastAPI与aiohttp实现SSE通信异常：连接立即断开求助

问题场景 #

使用FastAPI搭建SSE（Server-Sent Events）服务端，aiohttp作为客户端发起请求后，返回200状态码但连接立即终止，服务端仅打印初始请求日志，客户端无法接收后续持续推送的数据。

核心问题分析 #

连接中断的主要原因包括：

• SSE消息格式不符合规范，客户端无法识别持续流
• 服务端未配置正确的响应头，导致中间件（如OpenResty）截断长连接
• 客户端GET请求携带JSON Body的方式违反HTTP规范
• 生成器中存在字节转字符串的错误，导致消息格式异常

具体修复步骤 #

1. 修复服务端SSE消息格式 #

修改messages_generator方法，确保消息符合SSE标准格式：

• 移除current_time的encode('utf-8')操作，直接生成字符串
• 明确用\n\n作为每条消息的结束标记（避免字符串换行歧义）
• 使用json.dumps()序列化字典，保证消息格式合法

import json
import time
import asyncio
import logging

async def messages_generator(self, request):
    self.accesskey = request["accesskey"]
    logging.info(f"accesskey: {self.accesskey}")
    # 发送连接确认消息，符合SSE格式
    yield f"data: {self.accesskey} is connected\n\n"
    while True:
        current_time = time.strftime("%Y-%m-%d %H:%M:%S")
        message = {"event": "time_data", "data": current_time}
        logging.info(f"accesskey: {self.accesskey}, message: {message}")
        # 序列化字典为JSON字符串，确保消息格式正确
        yield f"data: {json.dumps(message)}\n\n"
        await asyncio.sleep(1)

2. 服务端添加SSE专属响应头 #

在返回StreamingResponse时，添加防止缓存、维持长连接的响应头：

from fastapi import FastAPI, StreamingResponse

app = FastAPI()

@app.get("/trader/MacroSSE")
async def trader_MacroSSE(item: traderlib.data_class.GeneralItem):
    accesskey_name = accesskey_chack(item)
    logger.info(f"accesskey_name: {accesskey_name}, item: {item}")
    
    # 配置SSE响应头
    headers = {
        "Cache-Control": "no-cache",
        "Connection": "keep-alive",
        "Content-Type": "text/event-stream",
        "Transfer-Encoding": "chunked"
    }
    # 如果item是Pydantic模型，转为字典传递给生成器
    return StreamingResponse(
        traderlib.MacroSSE().messages_generator(item.dict()),
        headers=headers,
        media_type="text/event-stream"
    )

3. 修复客户端请求方式 #

GET请求不应携带JSON Body，改为通过查询字符串传递参数：

import aiohttp
import asyncio

start_data = {"accesskey": "*****"}

async def sse_client(url, retry_interval=5):
    async with aiohttp.ClientSession() as session:
        while True:
            try:
                # 将参数作为查询字符串传递，符合GET请求规范
                async with session.get(url, params=start_data) as response:
                    response.raise_for_status()
                    print(response)

                    async for line in response.content:
                        # 解码字节为字符串，过滤空行
                        decoded_line = line.decode('utf-8').strip()
                        if decoded_line:
                            print(decoded_line)

            except (
                aiohttp.ClientError,
                aiohttp.http_exceptions.HttpProcessingError,
            ) as e:
                print(f"Network error: {e}")
            except asyncio.TimeoutError:
                print("Timeout error")
            except Exception as e:
                print(f"Unexpected error: {e}")

            print(f"Reconnecting in {retry_interval} seconds...")
            await asyncio.sleep(retry_interval)

url = "*****"
asyncio.run(sse_client(url))

4. 中间件配置（如OpenResty/Nginx） #

如果服务端部署在反向代理后，需禁用缓存和缓冲，维持长连接：

location /trader/MacroSSE {
    proxy_pass http://your-fastapi-server;
    proxy_set_header Connection '';
    proxy_http_version 1.1;
    chunked_transfer_encoding off;
    proxy_buffering off;
    proxy_cache off;
}

验证效果 #

修复完成后：

• 服务端日志会每秒打印一条推送消息
• 客户端会持续收到类似以下的输出：

  data: {"event": "time_data", "data": "2024-01-24 22:00:00"}
  data: {"event": "time_data", "data": "2024-01-24 22:00:01"}
  

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