音视频系列——Websockets接口封装为Http接口

模子服务示例：实时语音转文本服务

本示例展示一个支持双协议（WebSocket流式接口+HTTP同步接口）的语音转文本模子服务，并提供将WebSocket接口封装为HTTP接口的代码实现。

---

一、服务架构计划

---

二、WebSocket流式接口实现（Python）

使用FastAPI实现流式语音辨认服务：

1. # websocket_server.py
2. from fastapi import FastAPI, WebSocket
3. import asyncio
5. app = FastAPI()
7. async def process_audio_stream(bytes_data: bytes) -> str:
8.     # 调用语音识别模型（示例用伪代码）
9.     return "识别文本片段"
11. @app.websocket("/stream/asr")
12. async def websocket_asr(websocket: WebSocket):
13.     await websocket.accept()
14.     try:
15.         while True:
16.             audio_chunk = await websocket.receive_bytes()
17.             text = await process_audio_stream(audio_chunk)
18.             await websocket.send_text(text)
19.     except WebSocketDisconnect:
20.         print("客户端断开连接")

复制代码

技能特点：
• 支持分片音频流实时辨认
• 全双工通讯低沉延伸
• 主动处理处罚毗连断开

---

三、HTTP同步接口实现

1. # http_server.py
2. from fastapi import FastAPI, File, UploadFile
3. import httpx
5. app = FastAPI()
7. @app.post("/api/asr")
8. async def http_asr(file: UploadFile = File(...)):
9.     # 调用内部WebSocket服务（后文实现封装）
10.     async with httpx.AsyncClient() as client:
11.         result = await client.post(
12.             "http://adapter:8000/adapt/asr",
13.             files={"file": (file.filename, await file.read())}
14.         )
15.     return result.json()

复制代码

---

四、WebSocket转HTTP适配器实现

方案1：异步署理模式（保举）

1. # adapter.py
2. from fastapi import FastAPI, UploadFile, File
3. import websockets
4. import asyncio
5. import uuid
7. app = FastAPI()
9. async def websocket_client(audio_data: bytes):
10.     async with websockets.connect("ws://localhost:8000/stream/asr") as ws:
11.         # 分片发送音频数据
12.         chunk_size = 1024
13.         for i in range(0, len(audio_data), chunk_size):
14.             await ws.send(audio_data[i:i+chunk_size])
15.             await asyncio.sleep(0.01)  # 模拟流式传输
16.         
17.         # 接收最终结果
18.         final_result = []
19.         while True:
20.             try:
21.                 result = await asyncio.wait_for(ws.recv(), timeout=1.0)
22.                 final_result.append(result)
23.             except (asyncio.TimeoutError, websockets.ConnectionClosed):
24.                 break
25.     return "".join(final_result)
27. @app.post("/adapt/asr")
28. async def adapt_http_to_ws(file: UploadFile = File(...)):
29.     audio_data = await file.read()
30.     return {"text": await websocket_client(audio_data)}

复制代码

方案2：消息队列桥接

1. # 使用Redis Stream实现
2. import redis
3. r = redis.Redis()
5. async def process_task(file_data: bytes):
6.     task_id = str(uuid.uuid4())
7.     # 将任务放入队列
8.     r.xadd("asr_tasks", {task_id: file_data})
9.    
10.     # 等待结果
11.     while True:
12.         result = r.get(f"result:{task_id}")
13.         if result:
14.             return result.decode()
15.         await asyncio.sleep(0.1)
17. @app.post("/queue/asr")
18. async def queue_adapter(file: UploadFile = File(...)):
19.     return {"text": await process_task(await file.read())}

复制代码

---

五、协议转换关键技能点

• 数据分片处理处罚
  • HTTP接口汲取完备文件后主动切分为WebSocket流式分片
  • 设置公道的数据块巨细（发起1-4KB）
  
• 超时控制
  1. # 设置10秒超时
  2. async with async_timeout.timeout(10):
  3.     return await websocket_client(data)

  复制代码
• 错误重试机制
  1. @retry(stop=stop_after_attempt(3), wait=wait_fixed(0.5))
  2. async def safe_websocket_call():
  3.     # 包含心跳检测的稳定连接

  复制代码
• 协议头转换
  1. # 携带HTTP认证头到WebSocket
  2. headers = {"Authorization": request.headers.get("Authorization")}
  3. async with websockets.connect(ws_url, extra_headers=headers) as ws:
  4.     # ...

  复制代码

---

六、性能对比

指标WebSocket流式接口HTTP封装接口延伸200-500ms1-2s吞吐量1000 req/s300 req/sCPU占用较高（连续毗连）较低（短毗连）实用场景实时语音/视频流文件上传/短文本开发复杂度必要处理处罚毗连状态简朴哀求相应模子

---

七、摆设发起

• 容器化设置
  1. # Dockerfile
  2. FROM python:3.9-slim
  3. RUN pip install fastapi uvicorn websockets redis
  4. EXPOSE 8000
  5. CMD ["uvicorn", "adapter:app", "--host", "0.0.0.0"]

  复制代码
• 负载平衡计谋
  1. # Nginx配置
  2. upstream asr_servers {
  3.     server ws1:8000;
  4.     server ws2:8000;
  5.     keepalive 10; # 保持WebSocket长连接
  6. }

  复制代码
• 监控指标
  • WebSocket毗连存活时间
  • HTTP哀求乐成率
  • 音频流分片处理处罚延伸
  

---

以上实现完备支持两种协议的混淆调用模式，开发者可根据现实场景选择适配方案。如需测试完备代码，发起参考WebSocket官方测试方法创建端到端验证流程。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。