随着AI对话、智能问答、大模型交互场景全面普及，打字机流式逐字输出效果已然成为AI产品的标配交互。区别于普通静态文本渲染，打字机动态输出能够模拟真人实时回复的视觉体验，大幅提升AI对话的流畅感与沉浸感，是前端、后端AI项目的高频刚需功能。在实际开发中，不同技术栈的实现逻辑、渲染机制、流式数据处理方式截然不同。为了适配不同项目技术架构，本文聚焦Vue2、Vue3、React三大前端框架，搭配Java、Python两大后端语言，全方位落地AI对话打字机效果。覆盖前端动态渲染、后端流式数据返回核心逻辑，对比各技术栈的实现思路、优缺点与适用场景，零基础可直接复用代码，快速适配大模型对话、智能客服、AI问答等各类实战项目。一、核心知识
1、前端3 种流式接收技术
（1）前端SSE（Server-Sent Events）
特点
最简代码（前端）
const eventSource = new EventSource("/api/stream");const div = document.getElementById("content");// 每次收到一段文字，直接追加 = 打字机效果eventSource.onmessage = (e) => { div.innerHTML += e.data; };// 结束关闭eventSource.onerror = () => eventSource.close();
优点
- 后端配合 SseEmitter / FastAPI 一行搞定
缺点
（2）前端Fetch 流式（ReadableStream）
原生 HTTP 流，手动解析 chunk，比 SSE 麻烦一点，但更灵活。特点
代码示例
async function start() { const res = await fetch("/api/stream"); const reader = res.body.getReader(); const decoder = new TextDecoder(); const div = document.getElementById("content"); while (true) { const { done, value } = await reader.read(); if (done) break; // 实时追加文字 = 打字机 div.innerHTML += decoder.decode(value); }}start();
优点
缺点
（3）前端WebSocket（不建议做实际项目使用）
特点
代码
const ws = new WebSocket("ws://localhost:8080/ws");ws.onmessage = (e) => { div.innerHTML += e.data;};
优点
缺点
2、后端4 种流式输出技术
（1）Java SseEmitter
Spring Boot 自带，专门配合 SSE 做流输出，完美打字机。最简代码
@GetMapping("/api/stream")public SseEmitter stream() { SseEmitter emitter = new SseEmitter(); new Thread(() -> { try { // 分段发文字 = 打字机 emitter.send("我"); Thread.sleep(100); emitter.send("是"); Thread.sleep(100); emitter.send("打"); emitter.send("字"); emitter.send("机"); emitter.complete(); } catch (Exception e) { emitter.completeWithError(e); } }).start(); return emitter;}
优点
（2）Java 原生流（HttpServletResponse）
不依赖 SseEmitter，纯原生输出流，超轻量。@GetMapping("/raw")publicvoidrawStream(HttpServletResponse response) throws Exception { response.setContentType("text/plain;charset=utf-8"); PrintWriter out = response.getWriter(); out.write("我"); out.flush(); // 必须 flush 才会实时推到前端 Thread.sleep(100); out.write("是"); out.flush(); // ...}
优点
缺点
（3）Python FastAPI 流
FastAPI 原生支持流式返回，Python 首选。from fastapi import FastAPIimport timefrom fastapi.responses import StreamingResponseapp = FastAPI()def generate(): text = "我是打字机效果" for c in text: yield c time.sleep(0.1)@app.get("/api/stream")def stream(): return StreamingResponse(generate())
优点
（4）Python Flask 流
Flask 也支持流式输出，比 FastAPI 稍麻烦。from flask import Flask, Responseimport timeapp = Flask(__name__)def generate(): for c in "我是打字机": yield c time.sleep(0.1)@app.route("/stream")def stream(): return Response(generate())
优点
3、技术选型总结
最优组合：前端SSE + Java SseEmitter 或 Python FastAPI Stream- 想最简单→ SSE + SseEmitter / FastAPI
- 想轻量无依赖→ Java 原生流 / Flask 流
二、实战示例
1、Vue2+Java
（1）页面Vue文件
<template> <mainclass="typewriter-page"> <h1class="typewriter-page-title">vue2+java的打字机效果</h1> <sectionclass="typewriter-demo"> <articlev-for="(demo, index) in demos":key="demo.mode.key"class="typewriter-card"> <h2class="typewriter-title">{{ demo.mode.label }}</h2> <divclass="typewriter-box"aria-live="polite"> <span>{{ demo.outputText || '点击按钮开始打字机效果' }}</span> <spanv-if="demo.isTyping"class="typewriter-cursor">|</span> </div> <buttonclass="typewriter-button"type="button":disabled="demo.isTyping" @click="startTypewriter(index)"> {{ demo.isTyping ? '打字中...' : `启动${demo.mode.label}` }} </button> <pv-if="demo.errorMessage"class="typewriter-error">{{ demo.errorMessage }}</p> </article> </section> </main></template><script>import { JAVA_CHAT_MODES, buildPromptRequestUrl, openSseTypewriterStream, readFetchTypewriterStream} from '@/api/typewriterStream'const createDemoState = mode => ({ mode, outputText: '', isTyping: false, errorMessage: '', abortController: null, eventSource: null})export default { name: 'TypewriterStreamDemo', data() { return { demos: JAVA_CHAT_MODES.map(createDemoState) } }, beforeDestroy() { // 离开页面时关闭还没结束的连接，避免切换菜单后后台继续接收流。 this.demos.forEach(demo => this.closeStream(demo)) }, methods: { closeStream(demo) { if (demo.abortController) { demo.abortController.abort() demo.abortController = null } if (demo.eventSource) { demo.eventSource.close() demo.eventSource = null } }, resetDemo(demo) { // 每次点击都从空文本开始，旧请求如果还在进行则先中断。 this.closeStream(demo) demo.outputText = '' demo.errorMessage = '' demo.isTyping = true }, appendChunk(demo, chunk) { // 每收到一个流式片段就追加到对应框里，形成打字机效果。 demo.outputText += chunk }, async startFetchTypewriter(demo, prompt) { // AbortController 让用户切换页面或再次触发时可以取消当前 Fetch 流。 const controller = new AbortController() demo.abortController = controller try { await readFetchTypewriterStream(buildPromptRequestUrl(demo.mode.url, prompt), { signal: controller.signal, onChunk: chunk => this.appendChunk(demo, chunk) }) } catch (error) { if (!controller.signal.aborted) { demo.errorMessage = error.message || 'Fetch 流式请求失败，请确认 Java 服务已启动。' } } finally { if (demo.abortController === controller) { demo.abortController = null } demo.isTyping = false } }, startSseTypewriter(demo, prompt) { // SSE 不走 fetch reader，直接监听后端按事件推送的文本片段。 demo.eventSource = openSseTypewriterStream(buildPromptRequestUrl(demo.mode.url, prompt), { onChunk: chunk => this.appendChunk(demo, chunk), onDone: () => { demo.isTyping = false demo.eventSource = null }, onError: error => { demo.errorMessage = error.message || 'SSE 流式请求失败，请确认 Java 服务已启动。' demo.isTyping = false demo.eventSource = null } }) }, startTypewriter(index) { const demo = this.demos[index] if (!demo || demo.isTyping) { return } // prompt 直接传当前组合名称，后端就能返回更自然的展示文案。 const prompt = `Vue2 + ${demo.mode.label}` this.resetDemo(demo) if (demo.mode.transport === 'sse') { this.startSseTypewriter(demo, prompt) return } void this.startFetchTypewriter(demo, prompt) } }}</script><stylescoped>.typewriter-page { min-height: 100svh; padding: 24px; background: #f6f8fb; box-sizing: border-box;}.typewriter-page .typewriter-page-title { width: min(1120px, 100%); margin: 0 auto 22px; color: #172033; font-size: 24px; font-weight: 700; line-height: 1.35; text-align: center; overflow-wrap: anywhere;}.typewriter-demo { display: grid; grid-template-columns: repeat(2, minmax(0, 1fr)); gap: 18px; width: min(1120px, 100%); margin: 0 auto;}.typewriter-card { min-width: 0; text-align: left;}.typewriter-page .typewriter-title { min-height: 44px; margin: 0 0 10px; color: #172033; font-size: 16px; font-weight: 700; line-height: 1.4; overflow-wrap: anywhere;}.typewriter-box { min-height: 180px; padding: 20px; color: #172033; font-size: 16px; line-height: 1.8; text-align: left; white-space: pre-wrap; word-break: break-word; background: #ffffff; border: 1px solid #d8e1ec; border-radius: 8px; box-sizing: border-box;}.typewriter-cursor { color: #0f766e; animation: cursor-blink 0.9s step-end infinite;}.typewriter-button { margin-top: 16px; min-height: 40px; padding: 0 18px; color: #ffffff; font-size: 14px; font-weight: 600; background: #142132; border: 1px solid #142132; border-radius: 8px; cursor: pointer;}.typewriter-button:disabled { cursor: not-allowed; opacity: 0.65;}.typewriter-error { margin: 12px 0 0; color: #dc2626; font-size: 13px;}@keyframes cursor-blink { 50% { opacity: 0; }}@media (max-width: 760px) { .typewriter-demo { grid-template-columns: 1fr; }}</style>
（2）Api接口JS文件
import { openTextEventStream, requestTextStream, resolveServiceBaseUrl } from '@/utils/request'const DEFAULT_JAVA_API_BASE_URL = 'http://localhost:9001'// Vue2/Vue3 统一访问 Spring Cloud Gateway，再由网关转发到 Java demo 服务。const javaApiBaseUrl = resolveServiceBaseUrl('VITE_JAVA_API_BASE_URL', DEFAULT_JAVA_API_BASE_URL)export const JAVA_CHAT_MODES = [ { key: 'fetch-native', label: 'Fetch + Java 原生流', transport: 'fetch', description: '前端使用 Fetch ReadableStream，后端使用 StreamingResponseBody 输出普通文本流。', url: `${javaApiBaseUrl}/api/typewriter/java/native-stream` }, { key: 'fetch-sse-emitter', label: 'Fetch + Java SseEmitter', transport: 'fetch', description: '前端依然使用 Fetch，但手动解析后端返回的 SSE 事件帧。', url: `${javaApiBaseUrl}/api/typewriter/java/sse-emitter` }, { key: 'sse-native', label: 'SSE + Java 原生流', transport: 'sse', description: '前端使用 EventSource，后端不用 SseEmitter，而是手写 SSE 帧。', url: `${javaApiBaseUrl}/api/typewriter/java/native-stream-sse` }, { key: 'sse-sse-emitter', label: 'SSE + Java SseEmitter', transport: 'sse', description: '前后端都采用标准 SSE 方式：前端 EventSource，后端 SseEmitter。', url: `${javaApiBaseUrl}/api/typewriter/java/sse-emitter` }]export function buildPromptRequestUrl(url, prompt) { // 同时兼容绝对地址和相对地址，方便独立运行或接入 qiankun 主应用。 const requestUrl = new URL(url, window.location.origin) requestUrl.searchParams.set('prompt', prompt) return /^https?:\/\//.test(url) ? requestUrl.toString() : `${requestUrl.pathname}${requestUrl.search}${requestUrl.hash}`}export async function readFetchTypewriterStream(url, { signal, onChunk }) { // Fetch 流式读取普通文本响应，每读到一个 chunk 就回调给页面追加显示。 return requestTextStream(url, { signal, onChunk })}export function openSseTypewriterStream(url, { onChunk, onDone, onError }) { // SSE 由浏览器 EventSource 维护长连接，后端发送 done 事件后主动关闭。 return openTextEventStream(url, { onChunk, onDone, onError })}
（3）Java接口
package com.microlink.demo.controller;import org.springframework.http.MediaType;import org.springframework.http.ResponseEntity;import org.springframework.http.HttpHeaders;import org.springframework.web.bind.annotation.CrossOrigin;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RequestParam;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RestController;import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBody;import java.util.ArrayList;import java.nio.charset.StandardCharsets;import java.util.List;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;/** * 打字机流式输出示例。 * * SseEmitter 用于浏览器 EventSource； * StreamingResponseBody 用于浏览器 Fetch ReadableStream。 */@CrossOrigin@RestController@RequestMapping("/api/typewriter/java")public class TypewriterStreamController { private static final long TYPEWRITER_DELAY_MILLIS = 42L; /** * SSE 示例：EventSource 会自动按 text/event-stream 协议解析 message 事件。 */ @GetMapping(path = "/sse-emitter", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter streamWithSseEmitter(@RequestParam(value = "prompt", required = false) String prompt) { // 0L 表示不主动超时，便于演示长连接持续推送。 SseEmitter emitter = new SseEmitter(0L); // SseEmitter 的发送动作放到独立线程里，避免阻塞 Web 请求线程。 ExecutorService executorService = Executors.newSingleThreadExecutor(); List<String> typewriterTokens = buildTypewriterTokens(prompt, "Java SseEmitter"); executorService.execute(() -> { try { for (String token : typewriterTokens) { // 每个 token 作为一个 message 事件发给 EventSource，前端收到后立即追加。 emitter.send(SseEmitter.event().name("message").data(token)); if (!sleepForTypewriter()) { emitter.complete(); return; } } // 发送 done 事件，让前端可以主动关闭 EventSource。 emitter.send(SseEmitter.event().name("done").data("[DONE]")); emitter.complete(); } catch (Exception exception) { emitter.completeWithError(exception); } finally { executorService.shutdown(); } }); return emitter; } /** * Java 原生流示例：StreamingResponseBody 会保持连接并不断 flush 小块文本。 */ @GetMapping(path = "/native-stream", produces = "text/plain;charset=UTF-8") public StreamingResponseBody streamWithNativeResponse( @RequestParam(value = "prompt", required = false) String prompt ) { List<String> typewriterTokens = buildTypewriterTokens(prompt, "Java 原生流"); return outputStream -> { for (String token : typewriterTokens) { outputStream.write(token.getBytes(StandardCharsets.UTF_8)); // 每写入一个 token 都 flush，让浏览器不用等响应结束就能读取到内容。 outputStream.flush(); if (!sleepForTypewriter()) { break; } } }; } /** * Java 原生流 + SSE 协议示例：不使用 SseEmitter，直接手写 event/data 帧给 EventSource 消费。 */ @GetMapping(path = "/native-stream-sse", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public ResponseEntity<StreamingResponseBody> streamWithNativeSse( @RequestParam(value = "prompt", required = false) String prompt ) { List<String> typewriterTokens = buildTypewriterTokens(prompt, "Java 原生流"); StreamingResponseBody stream = outputStream -> { for (String token : typewriterTokens) { outputStream.write(buildSseEventFrame("message", token).getBytes(StandardCharsets.UTF_8)); outputStream.flush(); if (!sleepForTypewriter()) { return; } } outputStream.write(buildSseEventFrame("done", "[DONE]").getBytes(StandardCharsets.UTF_8)); outputStream.flush(); }; return ResponseEntity.ok() // EventSource 只接受 text/event-stream，这里显式写死响应头，避免被默认值覆盖成 text/plain。 .header(HttpHeaders.CACHE_CONTROL, "no-cache") .contentType(MediaType.TEXT_EVENT_STREAM) .body(stream); } private static List<String> buildTypewriterTokens(String prompt, String transportLabel) { String normalizedPrompt = normalizePrompt(prompt); String replyText = String.format( "当前展示是“%s”的打字机效果。当前响应来自%s，文本会按连续小片段逐步推送到前端，所以页面内容会一段一段显示出来。", normalizedPrompt, transportLabel ); List<String> tokens = new ArrayList<>(replyText.length()); // 示例里按字符拆分，真实业务可以换成大模型 token 或任务进度片段。 for (int index = 0; index < replyText.length(); index++) { tokens.add(String.valueOf(replyText.charAt(index))); } return tokens; } private static String buildSseEventFrame(String eventName, String data) { return "event: " + eventName + "\n" + "data: " + data + "\n\n"; } private static String normalizePrompt(String prompt) { if (prompt == null || prompt.trim().isEmpty()) { return "流式打字机效果"; } return prompt.trim(); } private static boolean sleepForTypewriter() { try { Thread.sleep(TYPEWRITER_DELAY_MILLIS); return true; } catch (InterruptedException exception) { // 恢复中断标记，交给容器或上层任务调度决定后续处理。 Thread.currentThread().interrupt(); return false; } }}
2、Vue3+Java
（1）页面Vue文件
<template> <mainclass="typewriter-page"> <h1class="typewriter-page-title">vue3+java的打字机效果</h1> <sectionclass="typewriter-demo"> <articlev-for="(demo, index) in demos":key="demo.mode.key"class="typewriter-card"> <h2class="typewriter-title">{{ demo.mode.label }}</h2> <divclass="typewriter-box"aria-live="polite"> <span>{{ demo.outputText || '点击按钮开始打字机效果' }}</span> <spanv-if="demo.isTyping"class="typewriter-cursor">|</span> </div> <buttonclass="typewriter-button"type="button":disabled="demo.isTyping" @click="startTypewriter(index)"> {{ demo.isTyping ? '打字中...' : `启动${demo.mode.label}` }} </button> <pv-if="demo.errorMessage"class="typewriter-error">{{ demo.errorMessage }}</p> </article> </section> </main></template><scriptsetuplang="ts">import { onBeforeUnmount, reactive } from 'vue'import { JAVA_CHAT_MODES, buildPromptRequestUrl, openSseTypewriterStream, readFetchTypewriterStream, type ChatMode} from '@/api/typewriterStream'type DemoState = { mode: ChatMode outputText: string isTyping: boolean errorMessage: string abortController: AbortController | null eventSource: EventSource | null}const demos = reactive<DemoState[]>( JAVA_CHAT_MODES.map(mode => ({ mode, outputText: '', isTyping: false, errorMessage: '', abortController: null, eventSource: null })))const closeStream = (demo: DemoState) => { demo.abortController?.abort() demo.abortController = null demo.eventSource?.close() demo.eventSource = null}const resetDemo = (demo: DemoState) => { // 每次启动前先清理旧连接，防止两个流同时往同一个框里写文本。 closeStream(demo) demo.outputText = '' demo.errorMessage = '' demo.isTyping = true}const appendChunk = (demo: DemoState, chunk: string) => { // 每收到一个流式片段就追加到对应框里，形成打字机效果。 demo.outputText += chunk}const startFetchTypewriter = async (demo: DemoState, prompt: string) => { // AbortController 用于页面卸载或重复触发时中断 Fetch ReadableStream。 const controller = new AbortController() demo.abortController = controller try { await readFetchTypewriterStream(buildPromptRequestUrl(demo.mode.url, prompt), { signal: controller.signal, onChunk: chunk => appendChunk(demo, chunk) }) } catch (error) { if (!controller.signal.aborted) { demo.errorMessage = error instanceof Error ? error.message : 'Fetch 流式请求失败，请确认 Java 服务已启动。' } } finally { if (demo.abortController === controller) { demo.abortController = null } demo.isTyping = false }}const startSseTypewriter = (demo: DemoState, prompt: string) => { // SSE 用 EventSource 接收 message 事件，后端发送 done 时结束本轮打字。 demo.eventSource = openSseTypewriterStream(buildPromptRequestUrl(demo.mode.url, prompt), { onChunk: chunk => appendChunk(demo, chunk), onDone: () => { demo.isTyping = false demo.eventSource = null }, onError: error => { demo.errorMessage = error.message || 'SSE 流式请求失败，请确认 Java 服务已启动。' demo.isTyping = false demo.eventSource = null } })}const startTypewriter = (index: number) => { const demo = demos[index] if (!demo || demo.isTyping) { return } // prompt 直接传当前组合名称，后端就能返回更自然的展示文案。 const prompt = `Vue3 + ${demo.mode.label}` resetDemo(demo) if (demo.mode.transport === 'sse') { startSseTypewriter(demo, prompt) return } void startFetchTypewriter(demo, prompt)}onBeforeUnmount(() => { // 微前端路由切换会卸载子应用，卸载前要释放仍在进行的流连接。 demos.forEach(closeStream)})</script><stylescoped>.typewriter-page { min-height: 100svh; padding: 24px; background: #f6f8fb; box-sizing: border-box;}.typewriter-page .typewriter-page-title { width: min(1120px, 100%); margin: 0 auto 22px; color: #172033; font-size: 24px; font-weight: 700; line-height: 1.35; text-align: center; overflow-wrap: anywhere;}.typewriter-demo { display: grid; grid-template-columns: repeat(2, minmax(0, 1fr)); gap: 18px; width: min(1120px, 100%); margin: 0 auto;}.typewriter-card { min-width: 0; text-align: left;}.typewriter-page .typewriter-title { min-height: 44px; margin: 0 0 10px; color: #172033; font-size: 16px; font-weight: 700; line-height: 1.4; overflow-wrap: anywhere;}.typewriter-box { min-height: 180px; padding: 20px; color: #172033; font-size: 16px; line-height: 1.8; text-align: left; white-space: pre-wrap; word-break: break-word; background: #ffffff; border: 1px solid #d8e1ec; border-radius: 8px; box-sizing: border-box;}.typewriter-cursor { color: #0f766e; animation: cursor-blink 0.9s step-end infinite;}.typewriter-button { margin-top: 16px; min-height: 40px; padding: 0 18px; color: #ffffff; font-size: 14px; font-weight: 600; background: #142132; border: 1px solid #142132; border-radius: 8px; cursor: pointer;}.typewriter-button:disabled { cursor: not-allowed; opacity: 0.65;}.typewriter-error { margin: 12px 0 0; color: #dc2626; font-size: 13px;}@keyframes cursor-blink { 50% { opacity: 0; }}@media (max-width: 760px) { .typewriter-demo { grid-template-columns: 1fr; }}</style>
（2）Api接口TS文件
import { openTextEventStream, requestTextStream, resolveServiceBaseUrl, type EventStreamHandlers, type StreamRequestOptions} from '@/utils/request'const DEFAULT_JAVA_API_BASE_URL = 'http://localhost:9001'export interface ChatMode { key: string label: string transport: 'fetch' | 'sse' description: string url: string}export type ReadFetchTypewriterOptions = StreamRequestOptionsexport type OpenSseTypewriterOptions = EventStreamHandlers// Vue3 示例只依赖 Java 后端，默认通过 Spring Cloud Gateway 的 9001 端口访问。const javaApiBaseUrl = resolveServiceBaseUrl('VITE_JAVA_API_BASE_URL', DEFAULT_JAVA_API_BASE_URL)export const JAVA_CHAT_MODES: ChatMode[] = [ { key: 'fetch-native', label: 'Fetch + Java 原生流', transport: 'fetch', description: '前端使用 Fetch ReadableStream，后端使用 StreamingResponseBody 输出普通文本流。', url: `${javaApiBaseUrl}/api/typewriter/java/native-stream` }, { key: 'fetch-sse-emitter', label: 'Fetch + Java SseEmitter', transport: 'fetch', description: '前端依然使用 Fetch，但手动解析后端返回的 SSE 事件帧。', url: `${javaApiBaseUrl}/api/typewriter/java/sse-emitter` }, { key: 'sse-native', label: 'SSE + Java 原生流', transport: 'sse', description: '前端使用 EventSource，后端不用 SseEmitter，而是手写 SSE 帧。', url: `${javaApiBaseUrl}/api/typewriter/java/native-stream-sse` }, { key: 'sse-sse-emitter', label: 'SSE + Java SseEmitter', transport: 'sse', description: '前后端都采用标准 SSE 方式：前端 EventSource，后端 SseEmitter。', url: `${javaApiBaseUrl}/api/typewriter/java/sse-emitter` }]export function buildPromptRequestUrl(url: string, prompt: string) { // 保留绝对 URL 用于跨服务访问；相对 URL 则交给当前前端站点代理或主应用处理。 const requestUrl = new URL(url, window.location.origin) requestUrl.searchParams.set('prompt', prompt) if (/^https?:\/\//.test(url)) { return requestUrl.toString() } return `${requestUrl.pathname}${requestUrl.search}${requestUrl.hash}`}export async function readFetchTypewriterStream(url: string, { signal, onChunk }: ReadFetchTypewriterOptions) { // Java 原生流返回 text/plain，前端通过 ReadableStream 分片读取。 return requestTextStream(url, { signal, onChunk })}export function openSseTypewriterStream(url: string, { onChunk, onDone, onError }: OpenSseTypewriterOptions) { // Java SseEmitter 返回 text/event-stream，EventSource 会按 message/done 事件分发。 return openTextEventStream(url, { onChunk, onDone, onError })}
（3）Java接口
同Vue2的Java接口
3、React+Python
（1）页面React的TSX文件
import { useEffect, useRef, useState } from 'react'import { PYTHON_CHAT_MODES, buildPromptRequestUrl, openSseTypewriterStream, readFetchTypewriterStream, type ChatMode,} from '@/api/typewriterStream'import './index.css'type DemoState = { mode: ChatMode outputText: string isTyping: boolean errorMessage: string}const createDemoStates = (): DemoState[] => PYTHON_CHAT_MODES.map(mode => ({ mode, outputText: '', isTyping: false, errorMessage: '', }))export default function TypewriterStreamPage() { const [demos, setDemos] = useState<DemoState[]>(createDemoStates) const abortControllersRef = useRef<Array<AbortController | null>>([]) const eventSourcesRef = useRef<Array<EventSource | null>>([]) const updateDemo = (index: number, patch: Partial<DemoState>) => { setDemos(currentDemos => currentDemos.map((demo, demoIndex) => (demoIndex === index ? { ...demo, ...patch } : demo)), ) } const appendChunk = (index: number, chunk: string) => { // 每收到一个流式片段就追加到对应框里，形成打字机效果。 setDemos(currentDemos => currentDemos.map((demo, demoIndex) => demoIndex === index ? { ...demo, outputText: `${demo.outputText}${chunk}` } : demo, ), ) } const closeStream = (index: number) => { abortControllersRef.current[index]?.abort() abortControllersRef.current[index] = null eventSourcesRef.current[index]?.close() eventSourcesRef.current[index] = null } const startFetchTypewriter = async (index: number, prompt: string) => { const demo = demos[index] if (!demo || demo.isTyping) { return } // 每个卡片维护自己的 AbortController，两个 Python 示例可以互不影响地取消。 const controller = new AbortController() abortControllersRef.current[index] = controller try { // 后端持续输出文本片段，onChunk 负责把片段追加到当前卡片的内容里。 await readFetchTypewriterStream(buildPromptRequestUrl(demo.mode.url, prompt), { signal: controller.signal, onChunk: chunk => appendChunk(index, chunk), }) } catch (error) { if (!controller.signal.aborted) { updateDemo(index, { errorMessage: error instanceof Error ? error.message : '流式请求失败，请确认 Python 服务已启动。', }) } } finally { if (abortControllersRef.current[index] === controller) { abortControllersRef.current[index] = null } updateDemo(index, { isTyping: false }) } } const startSseTypewriter = (index: number, prompt: string) => { const eventSource = openSseTypewriterStream(buildPromptRequestUrl(demos[index].mode.url, prompt), { onChunk: chunk => appendChunk(index, chunk), onDone: () => { eventSourcesRef.current[index] = null updateDemo(index, { isTyping: false }) }, onError: error => { eventSourcesRef.current[index] = null updateDemo(index, { errorMessage: error.message || 'SSE 请求失败，请确认 Python 服务已启动。', isTyping: false, }) }, }) eventSourcesRef.current[index] = eventSource } const startTypewriter = async (index: number) => { const demo = demos[index] if (!demo || demo.isTyping) { return } closeStream(index) const prompt = `React + ${demo.mode.label}` updateDemo(index, { outputText: '', errorMessage: '', isTyping: true, }) if (demo.mode.transport === 'sse') { startSseTypewriter(index, prompt) return } await startFetchTypewriter(index, prompt) } useEffect(() => { const abortControllers = abortControllersRef.current const eventSources = eventSourcesRef.current return () => { // qiankun 切走 React 子应用时会卸载组件，这里统一取消未结束的流请求。 abortControllers.forEach(controller => controller?.abort()) eventSources.forEach(eventSource => eventSource?.close()) } }, []) return ( <mainclassName="typewriter-page"> <h1className="typewriter-page-title">react+python的打字机效果</h1> <sectionclassName="typewriter-demo"> {demos.map((demo, index) => ( <articleclassName="typewriter-card"key={demo.mode.key}> <h2className="typewriter-title">{demo.mode.label}</h2> <divclassName="typewriter-box"aria-live="polite"> <span>{demo.outputText || '点击按钮开始打字机效果'}</span> {demo.isTyping ? <spanclassName="typewriter-cursor">|</span> : null} </div> <button className="typewriter-button" type="button" disabled={demo.isTyping} onClick={() => { void startTypewriter(index) }} > {demo.isTyping ? '打字中...' : `启动${demo.mode.label}`} </button> {demo.errorMessage ? <pclassName="typewriter-error">{demo.errorMessage}</p> : null} </article> ))} </section> </main> )}
（2）页面的CSS文件
.typewriter-page { min-height: 100svh; padding: 24px; background: #f6f8fb; box-sizing: border-box;}/* 提高选择器权重，避免被主应用对子应用 h1 的通用样式覆盖。 */main.typewriter-page > h1.typewriter-page-title { width: min(1120px, 100%); margin: 0 auto 22px; color: #172033; font-size: 24px; font-weight: 700; line-height: 1.35; text-align: center; overflow-wrap: anywhere;}.typewriter-demo { display: grid; grid-template-columns: repeat(2, minmax(0, 1fr)); gap: 18px; width: min(1120px, 100%); margin: 0 auto;}.typewriter-card { min-width: 0; text-align: left;}/* 卡片标题也固定为页面自己的尺寸，不继承主应用的 h2 放大样式。 */main.typewriter-page .typewriter-card > h2.typewriter-title { min-height: 44px; margin: 0 0 10px; color: #172033; font-size: 16px; font-weight: 700; line-height: 1.4; overflow-wrap: anywhere;}.typewriter-box { min-height: 180px; padding: 20px; color: #172033; font-size: 16px; line-height: 1.8; text-align: left; white-space: pre-wrap; word-break: break-word; background: #ffffff; border: 1px solid #d8e1ec; border-radius: 8px; box-sizing: border-box;}.typewriter-cursor { color: #0f766e; animation: cursor-blink 0.9s step-end infinite;}.typewriter-button { margin-top: 16px; min-height: 40px; padding: 0 18px; color: #ffffff; font-size: 14px; font-weight: 600; background: #142132; border: 1px solid #142132; border-radius: 8px; cursor: pointer;}.typewriter-button:disabled { cursor: not-allowed; opacity: 0.65;}.typewriter-error { margin: 12px 0 0; color: #dc2626; font-size: 13px;}@keyframes cursor-blink { 50% { opacity: 0; }}@media (max-width: 760px) { .typewriter-demo { grid-template-columns: 1fr; }}
（3）Api接口TS文件
import { openTextEventStream, requestTextStream, resolveServiceBaseUrl, type EventStreamHandlers, type StreamRequestOptions,} from '@/utils/request'const DEFAULT_FASTAPI_BASE_URL = 'http://localhost:8000'const DEFAULT_FLASK_API_BASE_URL = 'http://localhost:8000'export interface ChatMode { key: string label: string transport: 'fetch' | 'sse' description: string url: string}export type ReadFetchTypewriterOptions = StreamRequestOptionsexport type OpenSseTypewriterOptions = EventStreamHandlers// Python 示例统一跑在一个 FastAPI 应用里，FastAPI 和 Flask 两种接口共用 8000 端口。const fastApiBaseUrl = resolveServiceBaseUrl('VITE_FASTAPI_BASE_URL', DEFAULT_FASTAPI_BASE_URL)const flaskApiBaseUrl = resolveServiceBaseUrl('VITE_FLASK_API_BASE_URL', DEFAULT_FLASK_API_BASE_URL)export const PYTHON_CHAT_MODES: ChatMode[] = [ { key: 'fetch-fastapi', label: 'Fetch + FastAPI 流式', transport: 'fetch', description: 'StreamingResponse 持续返回 chunk，适合 React 的 Fetch 流式对话。', url: `${fastApiBaseUrl}/api/typewriter/fastapi/stream` }, { key: 'sse-fastapi', label: 'SSE + FastAPI 流式', transport: 'sse', description: 'FastAPI 也可以直接输出 SSE 帧，前端用 EventSource 持续接收。', url: `${fastApiBaseUrl}/api/typewriter/fastapi/sse` }, { key: 'fetch-flask', label: 'Fetch + Flask 流式', transport: 'fetch', description: 'Flask 生成器响应逐段输出，方便观察最轻量的 Python 流式实现。', url: `${flaskApiBaseUrl}/api/typewriter/flask/stream` }, { key: 'sse-flask', label: 'SSE + Flask 流式', transport: 'sse', description: 'Flask 也可以逐帧输出标准 SSE，前端直接交给 EventSource 处理。', url: `${flaskApiBaseUrl}/api/typewriter/flask/sse` }]export function buildPromptRequestUrl(url: string, prompt: string) { // 支持独立运行时的绝对 URL，也支持被主应用代理后的相对 URL。 const requestUrl = new URL(url, window.location.origin) requestUrl.searchParams.set('prompt', prompt) if (/^https?:\/\//.test(url)) { return requestUrl.toString() } return `${requestUrl.pathname}${requestUrl.search}${requestUrl.hash}`}export async function readFetchTypewriterStream(url: string, { signal, onChunk }: ReadFetchTypewriterOptions) { // React 页面两种 Python 后端都用 Fetch ReadableStream 接收文本分片。 return requestTextStream(url, { signal, onChunk })}export function openSseTypewriterStream(url: string, { onChunk, onDone, onError }: OpenSseTypewriterOptions) { // Python 端返回标准 text/event-stream 后，浏览器会自动分发 message / done 事件。 return openTextEventStream(url, { onChunk, onDone, onError })}
（3）Python接口
main.py
import uvicornPORT = 8000def run_app() -> None: """启动单个 Python 应用，FastAPI 和 Flask 示例接口共用同一个端口。""" print(f"FastAPI 流式服务：http://localhost:{PORT}/api/typewriter/fastapi/stream") print(f"FastAPI SSE 服务：http://localhost:{PORT}/api/typewriter/fastapi/sse") print(f"Flask 流式服务：http://localhost:{PORT}/api/typewriter/flask/stream") print(f"Flask SSE 服务：http://localhost:{PORT}/api/typewriter/flask/sse") uvicorn.run("app.fastapi_app:app", host="0.0.0.0", port=PORT)if __name__ == "__main__": run_app()
flask_app.py
from flask import Flask, Response, request, stream_with_contextfrom app.streaming.typewriter import iter_sse_typewriter_events, iter_typewriter_tokensdef create_app() -> Flask: app = Flask(__name__) @app.after_request def add_cors_headers(response: Response) -> Response: # 本地演示需要允许 React dev server 跨端口请求 Flask 流接口。 response.headers["Access-Control-Allow-Origin"] = "*" response.headers["Access-Control-Allow-Methods"] = "GET, OPTIONS" response.headers["Access-Control-Allow-Headers"] = "*" return response @app.get("/api/typewriter/flask/stream") def stream_typewriter() -> Response: """Flask 流式接口：生成器每 yield 一段文本，浏览器就能读到一段。""" prompt = request.args.get("prompt") return Response( # stream_with_context 让生成器执行期间仍能安全访问 Flask 请求上下文。 stream_with_context(iter_typewriter_tokens(prompt=prompt, backend_label="Python Flask 流")), mimetype="text/plain; charset=utf-8", ) @app.get("/api/typewriter/flask/sse") def stream_typewriter_sse() -> Response: """Flask SSE 接口：EventSource 直接按 message / done 事件消费。""" prompt = request.args.get("prompt") return Response( stream_with_context(iter_sse_typewriter_events(prompt=prompt, backend_label="Python Flask 流")), mimetype="text/event-stream; charset=utf-8", headers={"Cache-Control": "no-cache"}, ) return app
fastapi_app.py
from fastapi import FastAPIfrom fastapi.middleware.cors import CORSMiddlewarefrom fastapi.responses import StreamingResponsefrom starlette.middleware.wsgi import WSGIMiddlewarefrom app.flask_app import create_appfrom app.streaming.typewriter import iter_sse_typewriter_events, iter_typewriter_tokensapp = FastAPI(title="MicroLink Python Typewriter Streaming Demo")# 示例接口直接允许本地前端跨端口访问；生产环境应收敛为明确的域名白名单。app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=False, allow_methods=["GET", "OPTIONS"], allow_headers=["*"],)@app.get("/api/typewriter/fastapi/stream")def stream_typewriter(prompt: str | None = None) -> StreamingResponse: """FastAPI 流式接口：Fetch 可以通过 ReadableStream 逐块读取响应内容。""" return StreamingResponse( # 生成器每 yield 一个字符，StreamingResponse 就把它作为响应片段写给浏览器。 iter_typewriter_tokens(prompt=prompt, backend_label="Python FastAPI 流"), media_type="text/plain; charset=utf-8", )@app.get("/api/typewriter/fastapi/sse")def stream_typewriter_sse(prompt: str | None = None) -> StreamingResponse: """FastAPI SSE 接口：EventSource 直接按 message / done 事件消费。""" return StreamingResponse( iter_sse_typewriter_events(prompt=prompt, backend_label="Python FastAPI 流"), media_type="text/event-stream; charset=utf-8", headers={"Cache-Control": "no-cache"}, )# 把 Flask 示例挂到同一个 8000 端口；路径仍由 Flask app 自己处理。app.mount("/", WSGIMiddleware(create_app()))
三、写在最后
1、SSE 实现代码最简，单向推送适配打字机场景，浏览器原生支持2、Fetch 流式灵活性更强，支持 POST 请求，需手动解析二进制数据流3、WebSocket 为双向通信协议，功能冗余，打字机场景不优先选用4、Java SseEmitter 框架封装完善，对接 SSE 稳定性高，开发便捷5、Java 原生流无额外依赖，体量最轻，连接管控相对薄弱6、FastAPI 流式写法简洁高效，Python 技术栈首选方案7、Flask 可实现流式输出，适配老旧项目，易用性略低于 FastAPI8、打字机效果最优组合为前端 SSE 搭配 Java SseEmitter 或 FastAPI 流如果本文对你有帮助，不妨点个赞，关注一下～欢迎在评论区留言交流，一起学习进步，共同成长！
夜雨聆风
