不使用EventSource的POST请求的SSE（Server-Sent Events）

最近在写一个类似于ChatGPT的流式输出的后端接口，查找资料以及控制台查看到这一类的开发都是使用的SSE。后端查资料一顿开搞，前端却犯了难…前端使用EventSource这个对象在MDN中的定义是只允许发送GET请求，而且还不能携带头部信息…后面查找资料发现可以这样使用，故翻译收藏起来了。

本文将解释如何使用 HTTP POST 请求从前端接收 SSE，EventSource 不支持该请求。大多数文章和文档都会告诉您需要 GET 请求才能接收事件，但事实并非如此。大多数 HTTP 库都会隐藏用于从 axios 等服务器接收响应的较低级别步骤。我们需要一个库，允许我们根据收到的每个网络请求（又称每个 SSE 消息）构建自己的响应。

What is SSE?

SSE（Server-Sent Events）是一种基于 HTTP 协议的服务器推送技术，它允许服务器实时向客户端推送数据，而无需客户端定时轮询或发送请求。SSE 的主要特点包括：

  • 单向通信：SSE 是从服务器向客户端的单向通信，客户端不需要发送任何请求就可以接收到服务器的推送数据。
  • 实时性：SSE 的推送数据是实时的，客户端可以立即收到最新的数据。
  • 可靠性：SSE 是基于 HTTP 协议的，因此可以使用 HTTP 的可靠性机制，例如重试、重连和错误处理。

在 ChatGPT 的 AI 生成内容中，SSE 技术可以用来实时向客户端推送新生成的对话内容。当用户发起对话后，ChatGPT 会不断地生成新的对话内容，并通过 SSE 技术将新的内容实时地推送给客户端，从而实现实时对话的效果。

使用 SSE 技术的好处是能够节省客户端的网络带宽和服务器的资源，因为客户端不需要定时发送请求，而是等待服务器推送数据。此外，SSE 还具有实时性和可靠性的优势，可以保证客户端能够实时接收到最新的数据，而且在网络故障或其他错误时能够自动重试和重连。

SSE 通信图

How it works?

  1. 首先，我们将使用 fetch 向 SSE 端点发出 POST 请求。该端点将设置为返回多条带有响应标头 ContentType: text/event-stream 的消息。
1
2
3
4
5
6
7
8
9
const response = await fetch('/sse', {
  method: 'POST',
  headers: {
    'Content-Type': 'text/event-stream'
  },
  body: {
    "user_id": 123
  }
})
  1. 创建一个 reader 实例以获取从服务器收到的每个网络请求。
1
2
3
4
5
// To recieve data as a string we use TextDecoderStream class in pipethrough
const reader = response.body.pipeThrough(new TextDecoderStream()).getReader()

// To receive data as byte array we call getReader() directrly
const reader = response.body.getReader();
  1. 创建一个循环以继续接收消息，直到触发完成信号为止。在此循环内，还可以使用 SSE 消息更新前端应用程序。
1
2
3
4
5
while (true) {
  const {value, done} = await reader.read();
  if (done) break;
  console.log('Received', value);
}
  1. 完整代码示例

从 POST 请求接收 SSE 消息的完整示例。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
const response = await fetch('/sse', {
    method: 'POST',
    headers: {
      'Content-Type': 'text/event-stream'
    },
    body: {
      "user_id": 123
    }
  })
const reader = response.body.pipeThrough(new TextDecoderStream()).getReader()
while (true) {
  const {value, done} = await reader.read();
  if (done) break;
  console.log('Received', value);
}

用于完成设置循环的示例 Express 框架示例。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
app.post('/completion', (req, res) => {
  res.setHeader('Cache-Control', 'no-cache')
  res.setHeader('Content-Type', 'text/event-stream')
  res.setHeader('Access-Control-Allow-Origin', '*')
  res.setHeader('Connection', 'keep-alive')
  res.flushHeaders() // flush the headers to establish SSE with client

  const response = openai.createCompletion({
    model: 'text-davinci-003',
    prompt: 'hello world',
    max_tokens: 100,
    temperature: 0,
    stream: true
  }, { responseType: 'stream' })

  response.then(resp => {
    resp.data.on('data', data => {
      const lines = data.toString().split('\n').filter(line => line.trim() !== '')
      for (const line of lines) {
        const message = line.replace(/^data: /, '')
        if (message === '[DONE]') {
          res.write('data: DONE\n\n')
          res.end()
          return
        }
        const parsed = JSON.parse(message)
        res.write(`data: ${parsed.choices[0].text}\n\n`)
      }
    })
  }).catch(err => {
    console.log(err)
  })
})