undici EventSource 使用与原理详解：在 Node.js 中实现 Server-Sent Events

后端网络通信【免费下载链接】undiciAn HTTP/1.1 client, written from scratch for Node.js项目地址https://gitcode.com/gh_mirrors/un/undici点击查看免费下载EventSource是 undici 提供的 WHATWG 规范 并结合仓库源码从构造参数、实例属性、事件模型、重连机制到流式解析原理带你完整掌握在 undici 中消费 SSE 流的能力。概述WHATWG 标准的 Server-Sent Events 客户端EventSource接口在 undici 中是一个EventTarget的子类见 lib/web/eventsource/eventsource.js它打开一条到服务器的持久 HTTP 连接服务端以text/event-stream内容类型持续推送事件客户端收到后即时派发事件连接不会被关闭。它是浏览器原生EventSource的 Node.js 等价实现遵循 WHATWG Server-Sent Events 处理模型。需要注意两点稳定性该接口标记为Stability: 1 - Experimental实验性从 v6.5.0 开始加入接口可能在后续版本中发生变化。首次构造警告第一次构造EventSource时会发出一次性ExperimentalWarning其code为UNDICI-ES源码见 eventsource.js 构造函数。全局安装该接口还会被安装到globalThis上即globalThis.EventSource可直接使用安装列表见 lib/global.js。最小用法如下import { EventSource } from undici const eventSource new EventSource(http://localhost:3000) eventSource.onmessage (event) { console.log(event.data) }构造函数new EventSource(url[, eventSourceInitDict])参数说明参数类型说明url{string|URL}事件流地址。相对 URL 会基于当前环境的基础 URL 解析。eventSourceInitDict{Object}可选初始化字典包含以下成员eventSourceInitDict支持以下选项withCredentials{boolean}为true时请求的 credentials 模式设为includeCORS attribute 状态设为use-credentials否则 credentials 模式为same-origin。默认false。dispatcher{Dispatcher}底层请求使用的 dispatcher。默认全局 dispatcher。Stability: 0 - Deprecated已弃用请改用node.dispatcher。node{Object}undici 对标准EventSourceInit字典的扩展dispatcher{Dispatcher}底层请求使用的 dispatcher。默认全局 dispatcher。reconnectionTime{number}断线后等待重新建立连接的时间毫秒。服务端可通过retry字段覆盖此值。默认3000。注意dispatcher选项标记为弃用Stability 0推荐使用node.dispatcher。从源码看构造函数会优先取node.dispatcher其次才是顶层dispatcher见 eventsource.jsreconnectionTime的默认值3000与 Chrome 一致Deno 使用 5000定义在 源码常量。构造时的请求行为创建EventSource实例后会立即开始连接url。请求以以下特征发出对应 构造函数源码请求头Accept: text/event-streamcache 模式为no-storeinitiator 类型为other请求由createPotentialCORSRequest构造模式为corscredentials 模式依据withCredentials在same-origin与include之间切换见 lib/web/eventsource/util.js。如果url无法解析会抛出SyntaxError类型的DOMException。仓库测试 test/eventsource/eventsource.js 验证了不传url抛TypeError、非法 URL 抛带Invalid URL消息的 DOMException。带withCredentials的示例import { EventSource } from undici const eventSource new EventSource(http://localhost:3000, { withCredentials: true })自定义 Dispatcher 注入请求头可以通过自定义Dispatcher控制底层请求例如在派发时追加请求头import { EventSource, Agent } from undici class CustomHeaderAgent extends Agent { dispatch (opts) { opts.headers[x-custom-header] hello world return super.dispatch(...arguments) } } const eventSource new EventSource(http://localhost:3000, { node: { dispatcher: new CustomHeaderAgent() } })在 构造函数源码 中this.#dispatcher被直接保存并在每次#connect()时传给底层fetching()因此自定义 dispatcher 会作用于每次连接含重连。实例属性eventSource.readyState{number}只读表示连接状态取值为以下常量之一EventSource.CONNECTING0——连接尚未建立或连接关闭后正在重新建立EventSource.OPEN1——连接已打开正在按收到顺序派发事件EventSource.CLOSED2——连接未打开且不再尝试重连。eventSource.url{string}只读事件流地址为基于环境基础 URL 解析后的最终 URL源码中保存的是new URL(url, baseUrl).href见 eventsource.js。eventSource.withCredentials{boolean}只读表示是否以 CORS credentials 方式实例化true或否false默认值反映构造时传入的withCredentials选项。事件处理器属性属性类型默认值触发时机eventSource.onopen{Function|null}null派发open事件时eventSource.onmessage{Function|null}null派发message事件时eventSource.onerror{Function|null}null派发error事件时赋值函数会注册为对应事件的处理器赋值为null则移除当前处理器。从源码实现看这些 setter 会先removeEventListener旧的处理器再通过webidl.converters.EventHandlerNonNull转换后addEventListener注册新处理器见 eventsource.js。静态常量CONNECTING/OPEN/CLOSEDEventSource.CONNECTING数值常量0EventSource.OPEN数值常量1EventSource.CLOSED数值常量2。它们被定义为只读且不可写的属性同时存在于EventSource构造函数与实例原型上通过Object.defineProperties以writable: false定义见 eventsource.js。方法eventSource.close()关闭连接如有中止底层请求并将readyState置为CLOSED。关闭后EventSource不会再尝试重连对已关闭的实例再次调用close()无任何效果。源码实现eventsource.js为将readyState置为CLOSED然后调用#controller.abort()中止 fetch并清空#request引用。另外#connect()与#reconnect()的入口都会检查readyState CLOSED并直接返回从而保证关闭后不再建立连接。事件Event:open连接建立、readyState变为OPEN时触发监听器收到一个Event对象。import { EventSource } from undici const eventSource new EventSource(http://localhost:3000) eventSource.addEventListener(open, () { console.log(connection opened) })Event:message当收到没有显式event字段的消息时触发v6.15.0 加入。监听器收到一个MessageEvent其data、lastEventId、origin属性由服务端事件填充。命名事件带event字段的事件会在自己的类型名下派发必须通过addEventListener()订阅。下面是一个完整的服务端推送 客户端消费示例import { createServer } from node:http import { EventSource } from undici const server createServer((request, response) { response.writeHead(200, { content-type: text/event-stream, cache-control: no-cache, connection: keep-alive }) response.write(event: ping\n) response.write(data: connected\n\n) const interval setInterval(() { response.write(data: ${Date.now()}\n\n) }, 1000) request.on(close, () clearInterval(interval)) }) server.listen(3000, () { const eventSource new EventSource(http://localhost:3000) // Named event, delivered under its own type. eventSource.addEventListener(ping, (event) { console.log(ping:, event.data) }) // Unnamed event, delivered as message. eventSource.onmessage (event) { console.log(message:, event.data) } })注意示例中服务端发送了event: ping命名事件与若干无event字段的默认事件命名事件只能通过addEventListener(ping, ...)收到默认事件则统一进入onmessage/message。这正是 WHATWG SSE 处理模型中的命名事件按自身类型派发规则。Event:error连接失败或中断时触发监听器收到一个Event对象。当失败可恢复时EventSource回到CONNECTING状态并在重连时间后重试当失败不可恢复时EventSource转为CLOSED且不再重连。v7.11.0 行为变更网络错误后不再重新建立连接而是直接关闭EventSource。import { EventSource } from undici const eventSource new EventSource(http://localhost:3000) eventSource.onerror () { if (eventSource.readyState EventSource.CLOSED) { console.log(connection closed) } else { console.log(reconnecting) } }结合 processResponse 源码 可以看到error事件的几种触发路径响应为中止的网络错误aborted→ 关闭连接并派发error不再重连响应为普通网络错误→ 调用#reconnect()进入重连流程并派发error响应状态码非 200或Content-Type的 essence 不是text/event-stream→ 关闭并派发error连接中途流式传输失败非正常中止→ 关闭并派发error见 pipeline 回调。连接生命周期与重连机制EventSource在每次建立连接时都会克隆原始请求再发起 fetch而不是复用同一请求对象。源码注释eventsource.js解释了原因fetch 会修改传入的请求URL 列表、重定向次数、响应污染、跨源重定向时的头复用会导致重连指向最后一次重定向目标并在累计 20 次重定向后永久失败因此cloneRequest()克隆请求、并单独拷贝urlList。所有重定向301/308 永久重定向与 302/307 临时重定向按相同方式处理。重连流程#reconnect 源码若readyState为CLOSED直接中止将readyState置为CONNECTING派发error事件等待reconnectionTime毫秒后若readyState仍为CONNECTING若本地记录有lastEventId则在请求头中设置Last-Event-ID先删除旧值再设置且会校验值合法性见 isValidHeaderValue 检查重新调用#connect()。两个值得注意的实现细节定时器使用setTimeout(...).unref()因此仅剩重连定时器在跑时不会阻止 Node.js 进程退出重连时间会与maxReconnectionTime 2 ** 31 - 1取最小值。源码注释指出超过该值的延迟包括Infinity会让 Node.js 在 1ms 后触发定时器从而把很长的重连时间变成重连风暴见 eventsource.js。retry字段对重连时间的覆盖服务端可在事件流中发送retry: 毫秒字段客户端解析后会更新当前reconnectionTime。这在 EventSourceStream.processEvent 中实现仅当retry值全部为 ASCII 数字时才生效parseInt解析非法值被忽略。Last-Event-ID断点续传服务端可在事件中发送id: 字符串字段客户端将其记录为lastEventId校验规则是不含U0000NULL 字符见 util.js。重连时该值会通过Last-Event-ID请求头发回服务端实现断点续传例如事件流中途断开后服务端可依据该头从断点继续推送。若某条事件流中携带的id含 NULL 字符该字段会被忽略。流式解析原理EventSourceStream收到合法响应后undici 将响应体通过pipeline接入EventSourceStream一个 object-mode 的Transform流见 eventsource-stream.js逐字节解析text/event-stream格式并把解析结果 push 为MessageEvent。解析器实现了 WHATWG 规范的完整规则parseLine 实现空行结束当前事件若事件有data则派发否则丢弃以:开头的行注释行忽略data:字段累计事件数据同一事件内多行data以\n连接见 源码仅有data字段的事件派发为message类型携带data、lastEventId、originevent:字段为事件命名派发时使用该名称作为事件类型processEventid:字段更新lastEventIdretry:字段更新重连时间BOM 处理流开头若为 UTF-8 BOMEF BB BF会自动剥离handleBOM行结束符兼容LF、CR与CRLF三种行结束方式。每个事件的MessageEvent还带有origin指向响应最终 URL 的 origin跨源重定向后 origin 会更新为最终 origin见 eventsource.js。配置 EventSource 事件大小上限EventSource 相关的限制可以在 dispatcher 上通过eventSource选项配置。具体到Client详见 Client 构造参数eventSource.maxEventSize{number}EventSource 消息允许的最大事件大小字节。设为0可禁用限制。默认buffer.kStringMaxLength。该选项在 DispatcherBase 中被接收并暴露为eventSourceOptions随后在建立连接时传入EventSourceStream见 eventsource.js。当单个事件累计数据超过上限时解析器会抛出EventSource message size exceeded错误错误对象带aborted false见 eventsource-stream.js并导致连接关闭、派发error事件。进阶场景将 EventSource 与 Dispatcher 生命周期配合由于EventSource可注入任意Dispatcher推荐通过node.dispatcher你可以复用已配置的Agent/Client例如配置了代理、TLS、连接池参数的实例来承载事件流请求避免使用全局默认 dispatcher通过自定义Agent.dispatch()拦截请求如追加鉴权头、统计埋点正如上文CustomHeaderAgent示例通过Client的eventSource.maxEventSize收紧或放宽事件大小限制。小结undici 的EventSource提供了一个与浏览器行为一致的 SSE 客户端自动重连含retry字段覆盖与Last-Event-ID断点续传、完整的事件派发模型open/message/error与命名事件、兼容各种行结束符与 BOM 的流式解析器以及可通过 dispatcher 定制请求与事件大小限制的能力。其行为细节均可对照 EventSource 官方文档、实现源码 与 测试用例 进一步验证与探索。赞分享后端网络通信【免费下载链接】undiciAn HTTP/1.1 client, written from scratch for Node.js项目地址https://gitcode.com/gh_mirrors/un/undici点击查看免费下载相关推荐Undici事件流处理EventSource与Server-Sent Events实战Undici事件流处理EventSource与Server Sent Events实战 你是否还在为Node.js应用中的实时数据传输烦恼长轮询效率低下W后端网络通信Oak框架中的Server-Sent Events(SSE)实现详解Oak框架中的Server Sent Events SSE 实现详解 什么是Server Sent Events Server Sent Events 简称SS后端GeoIP2 Java API完整指南轻松实现IP地理位置查询的7个关键步骤GeoIP2 Java API完整指南轻松实现IP地理位置查询的7个关键步骤 GeoIP2 Java API是一个功能强大的开源库专门用于在Java应用中实后端上一篇免费离线跨平台drawio-desktop 流程图桌面工具完整指南下一篇RevokeMsgPatcher 完整实测PC 微信 QQ 防撤回补丁 4 分钟装完撤回彻底失效创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考