前言在进行直播数据分析、舆情研究或用户互动行为研究时弹幕数据是一类非常重要的实时文本数据来源。相比评论数据弹幕具有两个明显特点一是实时性强几乎与直播内容同步出现二是互动密度高能够反映观众情绪变化与热点话题。因此如果能够稳定地采集直播弹幕数据就可以进一步开展诸如情感分析、关键词统计、热点时刻识别等研究。在实际开发过程中我尝试设计了一套基于 WebSocket 的实时弹幕采集程序。整体思路并不复杂通过 WebSocket 服务接收弹幕消息对消息进行解析与格式化处理然后通过队列进行缓存并最终保存到日志文件和 JSON 文件中。这样不仅可以实现实时监控还能够方便后续进行数据分析与建模。本文结合具体代码分享这一套弹幕采集程序的实现思路与关键技术。一、环境准备与依赖安装在开始之前需要安装 WebSocket 相关依赖库。pip install websockets代码中还使用了以下 Python 内置库asyncio异步编程json数据解析logging日志系统os文件操作datetime时间处理导入模块如下import asyncio import json import websockets from collections import deque import logging import os from datetime import datetime这里特别需要注意的是asyncio 与 websockets的组合它能够实现异步消息处理非常适合实时弹幕这种高频数据流场景。二、日志系统与数据缓存设计在实际采集弹幕时如果只是在终端输出数据很容易丢失重要信息。因此我在代码中设计了日志系统 数据缓存队列。首先配置日志系统logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, datefmt%H:%M:%S )这样可以让每一条弹幕都带有时间信息方便后续查看与调试。接下来创建一个弹幕队列danmu_queue deque(maxlen100)这里使用deque的原因是插入速度快、支持固定长度缓存、适合实时数据流。三、弹幕数据存储结构设计为了便于后续分析我将弹幕数据保存为两种格式1️⃣ 文本日志2️⃣ JSON数据配置代码如下DATA_DIR danmu_data LOG_FILE os.path.join(DATA_DIR, danmu.log) JSON_FILE os.path.join(DATA_DIR, danmu.json) os.makedirs(DATA_DIR, exist_okTrue)这样程序运行时会自动创建一个danmu_data文件夹用来存储采集的数据.四、弹幕消息解析函数核心逻辑是process_message()函数它负责解析 WebSocket 接收到的消息。async def process_message(message_data):首先解析 JSON 数据data json.loads(message_data)接着判断消息类型if data.get(type) ! danmu: return None这里的设计非常重要因为 WebSocket 可能会发送多种类型的消息而我们只需要弹幕类型的数据。然后提取关键字段content message.get(content, ) sender message.get(nickname, unknown) time message.get(time, ) user_token message.get(userToken, ) live_id message.get(liveId, )最终构造一条格式化弹幕formatted_message f[{time}] {sender} [{user_token}]\n{content}例如[20:35:12] 用户12345 [token] 这主播太搞笑了随后将弹幕保存到字典中danmu { user: sender, content: content, time: time, live_id: live_id, user_token: user_token, formatted: formatted_message }并放入队列danmu_queue.append(danmu)最后调用函数保存到文件。五、WebSocket服务器实现为了接收弹幕数据需要搭建一个 WebSocket 服务。核心函数async def websocket_handler(websocket, path):当有客户端连接时client_id id(websocket) logging.info(f新的客户端连接 (ID: {client_id}))然后持续接收消息async for message in websocket: success await process_message(message)如果消息处理失败logging.warning(f消息处理失败: {message})当客户端断开连接时except websockets.exceptions.ConnectionClosed: logging.info(f客户端断开连接)这一部分逻辑实现了实时弹幕监听机制。六、服务器启动逻辑服务器入口函数async def main():启动 WebSocket 服务server await websockets.serve( websocket_handler, 127.0.0.1, 8765, ping_intervalNone )参数解释关闭 ping 的原因是某些客户端在频繁 ping 时可能会出现连接异常。七、弹幕数据保存机制为了方便数据分析代码设计了一个保存函数def save_danmu_to_file(danmu):首先写入文本日志with open(LOG_FILE, a, encodingutf-8) as f: f.write(danmu[formatted] \n\n)再写入 JSON 数据json.dump(danmu_with_timestamp, f, ensure_asciiFalse)并增加保存时间danmu_with_timestamp[save_time] datetime.now().isoformat()最终每条弹幕都会以 JSON 形式存储。例如{ user:张三, content:主播太厉害了, time:20:35:12, live_id:123456, user_token:abcde, save_time:2026-03-09T20:35:12 }这样既可以用于日志查看也可以直接用于数据分析。八、程序启动与关闭主程序入口if __name__ __main__:程序启动时会记录日志 程序启动于: 2026-03-09 20:30:01 程序结束时也会记录退出时间。这一步设计的好处是可以知道采集时长也可以追踪异常退出。九、弹幕采集程序的设计心得在设计这个弹幕采集程序时我有几个比较深的体会第一WebSocket非常适合实时数据采集。相比传统HTTP轮询方式WebSocket能够持续保持连接从而实时接收弹幕消息大幅降低延迟。第二异步编程是处理高频消息流的关键。使用asyncio可以避免阻塞提高程序整体效率。第三数据存储需要兼顾实时与分析需求。因此同时保存为日志文件和 JSON 文件这样既能实时查看也方便后续做数据分析。第四缓存队列设计可以提高系统稳定性。通过deque保存最近弹幕可以避免高频消息导致系统压力过大。十、总结通过这套程序我们实现了一套完整的直播弹幕实时采集系统具备以下特点1WebSocket实时采集2asyncio异步处理3弹幕队列缓存4自动日志记录5JSON结构化数据保存6程序运行状态记录这套程序不仅可以用于直播弹幕采集还可以扩展到实时舆情监测、弹幕情感分析、用户互动研究、直播热点识别。后续如果需要还可以进一步加入数据库存储MySQL / MongoDB、弹幕情感分析、实时数据可视化、弹幕关键词统计。从整体实现来看这套弹幕采集程序虽然代码量不算很大但在设计上我更关注的是稳定性和可扩展性。通过 WebSocket 实现实时数据接收再结合 asyncio 异步处理可以比较轻松地应对高频弹幕消息同时利用 deque 做缓存队列、日志系统做运行记录、JSON 做结构化数据存储使得整个数据流从“接收 → 解析 → 缓存 → 落盘”形成一个完整闭环。这样的设计不仅方便后续做弹幕文本分析、情感分析或关键词统计也为进一步接入数据库、实时可视化或数据流处理打下了基础。大家有任何想法也欢迎随时在评论区交流哦~