ServerSentEvent学习 | Mox的笔记库

起因

我记得我在高中时候问过老师：发送一次HTTP请求，能否收到多次HTTP请求回复。

老师的回答是：不能

直到我上了大学，2023年，不少ChatGPT聊天框的实现开始使用Server Sent Event（SSE）实现打字机样式的输出，我发现高中时的需求居然是可以实现的。在此之前，这一类需求我都是用WebSockets实现。Python的Flask和Django关于WebSockets属实让人不舒服，SpringBoot有关Websockets的配置也让我鼓弄了老半天（Java用的少是原罪）。但如果用SSE，回复只要按流式传输返回即可，个人感觉在理解上相对容易一些。

参考资料

MDN里面关于实现有很详细的说明

Using server-sent events

阮一峰的教程，一份很好的普及资料

Server-Sent Events 教程

学校挑战杯实践记录

在实践中发现：如果要从网页前端（通过摄像头录制视频）往后端传输视频，浏览器不提供视频压缩方案，且要么只能往后端通过流式传输每帧图片的Base64，要么就是通过Websockets传输每帧图片的Base64（本质上还是流式传输），这就让人很无语。

当时的应用场景是：后端处理前端发来的图片，对每张图片进行解析后发回解析的评价与解析过后的图片。我给出的解决方案是：视频/图片通过Websockets传输，图片解析通过SSE进行返回。

前端通过MediaRecorder API调用摄像头。mediaRecorder.start() 括号里面要填上间隔时间的，所以MDN的这个Example看着让人很迷惑（即Example直接运行是错误的）

而后端的Python，则需要利用Python yield的特性进行数据返回。如果不熟悉Python async操作的话也很容易踩上坑。

Python 异步迭代器解决TypeError: ‘async_generator‘ object is not iterable

在编写Python脚本对后端测试的时候，发现Python httpx的session居然无法复用，或者说复用后无法异步请求

浅度测评：requests、aiohttp、httpx 我应该用哪一个？

在Python中，有关Queue的nowait_开头的操作，在异步的情况下容易出错，但直接get和put就会直接锁死。难道Queue就没有加锁操作吗？

到了后面，项目要接入ChatGPT（图片的评价通过ChatGPT进行润色），于是也研究了一下ChatGPT的SSE返回。然而ChatGPT的SSE请求是通过POST请求开启的，而浏览器自带的EventSource API只支持GET请求。遇到这种情况就只能通过对Fetch API进行包装后进行实践。由于本人对JS/TS也只是个三脚猫的水准，外加项目急着答辩，自己写的封装死活写不对。最后用的是微软的封装@microsoft/fetch-event-source

在Fetch中使用SSE，支持POST

import { fetchEventSource } from '@microsoft/fetch-event-source';

对于OpenAI接口的实现样例

export function send1(){
    fetchEventSource(GPT, {
        method: 'POST',
        headers: {
            'Content-Type': 'application/json',
            'Authorization': "Bearer",
        },
        body: JSON.stringify({
            model: "gpt-3.5-turbo",
            stream: true,
            messages: [
                {"role": "system", "content": "你是个阳光开朗大男孩"},
                {"role": "user", "content": "你好" }
            ],
        }),
        onmessage(ev) {
            try{
                const result = JSON.parse(ev.data)
                console.log(ev.data)
            } catch (e) {
                console.log('解析 JSON 数据时出错：', e);
            }
        }
    });
}

Python Sanic框架有关SSE的Example

@app.route('/comment')
async def test(request):
    response = await request.respond(content_type='text/event-stream;charset=utf-8')
    await response.send("retry: 10000\n")
    await response.send("event: connecttime\n")
    await response.send("data: " + "hello" + "\n\n")
    while True:
        if not commentQueue.empty():
            await response.send("data: " + commentQueue.get_nowait() + "\n\n")
            await asyncio.sleep(1)
    await response.eof()

在前端，要想把摄像头的stream拿出来，就必须async getMedia

async function getMedia() {
    const constraints = {
        video: {width: 1080, height: 720},
        audio: false
    };
    stream = await navigator.mediaDevices.getUserMedia(constraints);
    video.value.srcObject = stream;
    video.value.play();
    // await video.value.play();
}
getMedia()

结语

通过学校挑战杯的这个项目算是对SSE技术有一个相对全面的了解，虽然中间也干出了不少大力出奇迹的事情（这样的事情在项目经常发生），但在录制/答辩现场顺利的跑了起来，也不算白忙活。记录一下，方便日后回顾。

推荐订阅源

Mox的笔记库

起因

参考资料

学校挑战杯实践记录

结语