为什么招聘高级前端开发这么难？
在现代前端开发中，使用Serve-Sent Events（SSE）实现流式数据传输（（如聊天消息、实时日志、AI 生成文本的逐字输出等））越来越流行，本文来介绍一个前端如何实现接受数据，解析数据，展示到页面，有很好的用户体验。
一、什么是Serve-Sent Events（SSE）？
@microsoft/fetch-event-source 是一个由微软开发的 JavaScript 库，旨在提供更灵活、功能更强大的服务器发送事件（Server-Sent Events, SSE）。它结合了浏览器原生的 fetch API 和 EventSource 的特性，允许开发者通过 HTTP 流（HTTP Streaming）实现实时数据传输，同时支持更多自定义配置（如请求头、身份认证、错误重试等）。
- 回调方法：
| 字段 | 含义 |
| method | 请求方法（POST、GET） |
| headers | 请求头Record<string, string>，通常需要指定'Content-Type': 'application/json','Accept': 'text/event-stream' |
| body | 请求的参数 |
| onopen | 响应回来的回调 |
| onmessage | 接收到消息的回调,每当服务器发一条消息，就触发接受一条消息 |
| onclose | 响应结束的回调 |
| onerror | 响应失败报错的回调 |
- 对比原生 API 的优势
| 特性 | @microsoft/fetch-event-source | 原生 EventSource |
| HTTP 方法 | 支持 GET/POST/PUT 等 | 仅 GET |
| 自定义请求头 | ✅ | ❌ |
| 请求体 | 支持任意数据（如 JSON） | 不支持 |
| 错误重试 | 可配置的重试逻辑 | 有限的重试 |
| 流控制 | 可手动暂停/恢复 | 不支持 |
| 页面隐藏时行为 | 可配置是否保持连接 | 默认暂停 |
适用场景推荐
- 需要与需要认证的 SSE 服务通信（如传递 Authorization 头）。
- 使用 POST 请求传递参数并接收流式响应（如 OpenAI 的流式 API）。
- 需要更健壮的连接管理和错误恢复机制。
二、如何使用@microsoft/fetch-event-source
首先，安装库 npm install @microsoft/fetch-event-source
安装成功，主要的基本用法
import { fetchEventSource } from '@microsoft/fetch-event-source';
async function startStream() {
await fetchEventSource('request url', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': 'Bearer YOUR_TOKEN',
},
body: JSON.stringify({ query: "Hello" }),
onopen(response) {
// 连接成功时触发
if (response.ok) return;
throw new Error('连接失败');
},
onmessage(event) {
// 接收服务器发送的每条事件
console.log('收到数据:', event.data);
// 请求完成
console.log('请求结束标记', data.done)
},
onclose() {
// 连接关闭时触发
console.log('连接终止');
},
onerror(err) {
// 错误处理（默认会抛出异常并自动重试）
console.error('错误:', err);
throw err; // 抛出错误会触发重试机制
}
});
}
三、在angular中实现，具体方案
1、安装库 npm install @microsoft/fetch-event-source
2、新建一个新的服务文件来处理 SSE 请求(chat.service.ts文件)
import { Injectable } from '@angular/core';
import { fetchEventSource } from '@microsoft/fetch-event-source';
import { Observable } from 'rxjs';
@Injectable({
providedIn: 'root'
})
export class ChatService {
streamGptResponse(prompt: string): Observable<string> {
return new Observable(observer => {
let fullResponse = '';
const ctrl = new AbortController();
fetchEventSource('request url', {
method: 'POST',
headers: {
'Content-Type': 'application/json', // 告诉服务器，客户端数据格式
'Accept': 'text/event-stream' // 客户端声明:希望接收事件流格式的响应
},
body: JSON.stringify({ prompt }),
signal: ctrl.signal,
onmessage(event) {
// 处理每个数据块
try {
const data = JSON.parse(event.data);
// data可以根据后端返回的具体格式，进行相应的处理
if (data.content) {
fullResponse += data.content;
observer.next(fullResponse);
}
// 标记请求完成
if (data.done) {
observer.complete();
}
} catch (error) {
console.error('Parse message failed:', error);
}
},
onopen(response) {
if (response.ok && response.headers.get('content-type')?.includes('text/event-stream')) {
console.log('Connection opened');
} else {
throw new Error(`Failed to open connection: ${response.status}`);
}
},
onclose() {
console.log('Connection closed');
observer.complete();
},
onerror(error) {
console.error('Connection error:', error);
observer.error(error);
ctrl.abort();
}
});
// 返回清理函数
return () => {
ctrl.abort();
};
});
}
}
3、创建一个组件使用该服务
安装markdown，解析样式: npm i ngx-markdown
推荐使用14.0.1版本，markdown的使用下一篇文章会解释
import { Component } from '@angular/core';
import { SseService } from '../../services/sse.service';
@Component({
selector: 'app-gpt-stream',
template: `
<div class="gpt-stream-container">
<div class="input-area">
<textarea
[(ngModel)]="prompt"
placeholder="请输入..."
></textarea>
<button
(click)="generateResponse()"
[disabled]="isGenerating"
>
发送
</button>
</div>
<div class="response-area">
<markdown *ngIf="response" [data]="response"></markdown>
<div *ngIf="isGenerating" class="loading">
思考中...
</div>
</div>
</div>
`,
styles: [`
.gpt-stream-container {
padding: 20px;
}
.input-area {
margin-bottom: 20px;
}
textarea {
width: 100%;
min-height: 100px;
padding: 10px;
margin-bottom: 10px;
}
.response-area {
padding: 15px;
border: 1px solid #eee;
border-radius: 4px;
min-height: 100px;
}
.loading {
color: #666;
font-style: italic;
}
`]
})
export class GptStreamComponent {
prompt = '';
response = '';
isGenerating = false;
constructor(private sseService: SseService) {}
generateResponse() {
if (!this.prompt.trim() || this.isGenerating) {
return;
}
this.isGenerating = true;
this.response = '';
this.sseService.streamGptResponse(this.prompt)
.subscribe({
next: (chunk) => {
this.response = chunk;
},
error: (error) => {
console.error('Stream error:', error);
this.isGenerating = false;
},
complete: () => {
this.isGenerating = false;
}
});
}
}
3、在 module 文件中注册组件和服务:
import { NgModule } from '@angular/core';
import { CommonModule } from '@angular/common';
import { FormsModule } from '@angular/forms';
import { MarkdownModule } from 'ngx-markdown';
import { GptStreamComponent } from './components/gpt-stream/gpt-stream.component';
import { SseService } from './services/sse.service';
@NgModule({
declarations: [
GptStreamComponent
],
imports: [
CommonModule,
FormsModule,
MarkdownModule.forRoot()
],
exports: [
GptStreamComponent
],
providers: [
SseService
]
})
export class SharedModule { }
六、总结
主要实现特点：
1. 流式处理:
- 使用 fetchEventSource 建立 SSE 连接
- 通过 Observable 包装事件流
- 实时更新 UI 显示生成的内容
2. 错误处理:
- 包含完整的错误处理机制
- 连接错误自动中断
- 解析错误的容错处理
3. 资源清理:
- 使用 AbortController 控制连接
- Observable 完成时自动清理
- 组件销毁时中断连接
4. 用户体验:
- 显示加载状态
- 防止重复提交
- 实时显示生成内容
5. 类型安全:
- 使用 TypeScript 类型
- 接口定义清晰
- 错误类型处理
原文链接：https://juejin.cn/post/7476881372810313762
