将 WebSocket 与 Python 结合使用

发布于2024-11-08

Using WebSocket with Python

什么是 WebSocket？

WebSocket 是一种支持浏览器和服务器之间实时、双向通信的协议。传统的 HTTP 通信涉及客户端发送请求和服务器响应以交换数据。相比之下，使用 WebSocket，一旦建立了初始连接，客户端和服务器都可以相互发送和接收消息，而无需重复建立新连接。

最近，OpenAI Realtime API 和 Hume AI 等交互式服务变得更加普遍，预计 WebSocket 的需求将会增加。本文介绍了如何使用 WebSocket 的基础知识，并介绍了相关的异步处理。

将 WebSocket 与 Python 结合使用

在Python中，可以使用WebSocket，如下所示：

import asyncio
import websockets

uri = "ws://..."

async def hello():
    async with websockets.connect(uri) as websocket:
        await websocket.send("Hello, Server!")
        response = await websocket.recv()
        print(f"Server says: {response}")

asyncio.run(hello())

使用 websockets.connect(uri) 连接到 WebSocket 服务器。
使用 websocket.send(message) 发送消息。
使用 websocket.recv() 接收消息。

异步处理

前面代码中使用的async和await表示异步处理。异步处理在同时执行多个任务时特别有效。

import asyncio

async def task1():
    print("Task 1: Start")
    await asyncio.sleep(2)  # Wait for 2 seconds
    print("Task 1: End")

async def task2():
    print("Task 2: Start")
    await asyncio.sleep(1)  # Wait for 1 second
    print("Task 2: End")

async def main():
    await asyncio.gather(task1(), task2())

asyncio.run(main())

在使用await的函数中，其他任务可以在等待当前任务完成的同时运行。这允许任务之间的高效切换。

异步处理和多线程

多线程也可以处理多个任务，但线程的利用方式有所不同：

在多线程中，每个任务都有自己的线程，程序在等待某些进程完成的同时在任务之间切换。
另一方面，异步处理不会创建新线程，而是在任务之间切换。

多线程在处理 CPU 密集型或阻塞操作时非常有效。然而，它也有缺点，例如线程切换（上下文切换）的开销和增加的内存消耗。

相反，异步处理减少了上下文切换的开销，因为它不依赖于线程。但是，如果正在运行繁重的任务，其他任务可能需要等待。因此，它适合 IO 密集型操作，例如 API 请求。

（对于计算密集型或需要精确计时的任务，多处理通常更有效。与多线程不同，多处理允许多个任务同时运行。）

例如，当使用OpenAI Realtime API实时接收来自麦克风的音频并将音频数据发送到API时，可以使用多线程和异步处理的组合：

import asyncio
import threading
import queue
import pyaudio
import websockets

# Use a queue to share data between threads
audio_queue = queue.Queue()

# Thread to capture audio using PyAudio
def audio_stream():
    p = pyaudio.PyAudio()
    stream = p.open(format=pyaudio.paInt16,
                    channels=1,
                    rate=44100,
                    input=True,
                    frames_per_buffer=1024)

    print("Start recording...")
    while True:
        data = stream.read(1024)
        audio_queue.put(data)

# Asynchronous function to send audio data via WebSocket
async def send_audio():
    uri = "ws://localhost:8765"
    async with websockets.connect(uri) as websocket:
        while True:
            # Get audio data from the queue
            data = audio_queue.get()
            if data is None:
                break
            await websocket.send(data)
            print("Sent audio data")

# Start the audio capture thread and run the asynchronous task
def main():
    audio_thread = threading.Thread(target=audio_stream)
    audio_thread.start()

    # Run the WebSocket sending task
    asyncio.run(send_audio())

if __name__ == "__main__":
    main()

音频捕获过程是一个阻塞操作，因此它是使用线程在单独的线程中执行的。相比之下，发送音频数据涉及 IO 绑定操作（例如与 API 交互），是使用异步处理完成的。（注意：PyAudio 也可以使用回调以非阻塞方式运行。）