前言

在進行直播數(shù)據(jù)分析、輿情研究或用戶互動行為研究時，彈幕數(shù)據(jù)是一類非常重要的實時文本數(shù)據(jù)來源。相比評論數(shù)據(jù)，彈幕具有兩個明顯特點：

一是實時性強，幾乎與直播內(nèi)容同步出現(xiàn)；

二是互動密度高，能夠反映觀眾情緒變化與熱點話題。因此，如果能夠穩(wěn)定地采集直播彈幕數(shù)據(jù)，就可以進一步開展諸如情感分析、關(guān)鍵詞統(tǒng)計、熱點時刻識別等研究。

在實際開發(fā)過程中，我嘗試設(shè)計了一套 基于 WebSocket 的實時彈幕采集程序。整體思路并不復(fù)雜：通過 WebSocket 服務(wù)接收彈幕消息，對消息進行解析與格式化處理，然后通過隊列進行緩存，并最終保存到日志文件和 JSON 文件中。這樣不僅可以實現(xiàn)實時監(jiān)控，還能夠方便后續(xù)進行數(shù)據(jù)分析與建模。

本文結(jié)合具體代碼，分享這一套彈幕采集程序的實現(xiàn)思路與關(guān)鍵技術(shù)。

一、環(huán)境準備與依賴安裝

在開始之前，需要安裝 WebSocket 相關(guān)依賴庫。

pip install websockets

代碼中還使用了以下 Python 內(nèi)置庫：

• asyncio（異步編程）
• json（數(shù)據(jù)解析）
• logging（日志系統(tǒng)）
• os（文件操作）
• datetime（時間處理）

導(dǎo)入模塊如下：

import asyncio
import json
import websockets
from collections import deque
import logging
import os
from datetime import datetime

這里特別需要注意的是 asyncio 與 websockets 的組合，它能夠?qū)崿F(xiàn) 異步消息處理，非常適合實時彈幕這種高頻數(shù)據(jù)流場景。

二、日志系統(tǒng)與數(shù)據(jù)緩存設(shè)計

在實際采集彈幕時，如果只是在終端輸出數(shù)據(jù)，很容易丟失重要信息。因此我在代碼中設(shè)計了 日志系統(tǒng) + 數(shù)據(jù)緩存隊列。

首先配置日志系統(tǒng)：

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    datefmt='%H:%M:%S'
)

這樣可以讓每一條彈幕都帶有時間信息，方便后續(xù)查看與調(diào)試。

接下來創(chuàng)建一個 彈幕隊列：

danmu_queue = deque(maxlen=100)

這里使用 deque 的原因是：插入速度快、支持固定長度緩存、適合實時數(shù)據(jù)流。

三、彈幕數(shù)據(jù)存儲結(jié)構(gòu)設(shè)計

為了便于后續(xù)分析，我將彈幕數(shù)據(jù)保存為兩種格式：

1?? 文本日志
2?? JSON數(shù)據(jù)

配置代碼如下：

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_ok=True)

這樣程序運行時會自動創(chuàng)建一個 danmu_data 文件夾，用來存儲采集的數(shù)據(jù).

四、彈幕消息解析函數(shù)

核心邏輯是 process_message() 函數(shù)，它負責解析 WebSocket 接收到的消息。

async def process_message(message_data):

首先解析 JSON 數(shù)據(jù)：

data = json.loads(message_data)

接著判斷消息類型：

if data.get('type') != 'danmu':
    return None

這里的設(shè)計非常重要，因為 WebSocket 可能會發(fā)送多種類型的消息，而我們只需要 彈幕類型的數(shù)據(jù)。

然后提取關(guān)鍵字段：

content = message.get('content', '')
sender = message.get('nickname', 'unknown')
time = message.get('time', '')
user_token = message.get('userToken', '')
live_id = message.get('liveId', '')

最終構(gòu)造一條格式化彈幕：

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)

最后調(diào)用函數(shù)保存到文件。

五、WebSocket服務(wù)器實現(xiàn)

為了接收彈幕數(shù)據(jù)，需要搭建一個 WebSocket 服務(wù)。

核心函數(shù)：

async def websocket_handler(websocket, path):

當有客戶端連接時：

client_id = id(websocket)
logging.info(f"新的客戶端連接 (ID: {client_id})")

然后持續(xù)接收消息：

async for message in websocket:
    success = await process_message(message)

如果消息處理失?。?/p>

logging.warning(f"消息處理失敗: {message}")

當客戶端斷開連接時：

except websockets.exceptions.ConnectionClosed:
    logging.info(f"客戶端斷開連接")

這一部分邏輯實現(xiàn)了 實時彈幕監(jiān)聽機制。

六、服務(wù)器啟動邏輯

服務(wù)器入口函數(shù)：

async def main():

啟動 WebSocket 服務(wù)：

server = await websockets.serve(
    websocket_handler,
    "127.0.0.1",
    8765,
    ping_interval=None
)

參數(shù)解釋：

關(guān)閉 ping 的原因是某些客戶端在頻繁 ping 時可能會出現(xiàn)連接異常。

七、彈幕數(shù)據(jù)保存機制

為了方便數(shù)據(jù)分析，代碼設(shè)計了一個保存函數(shù)：

def save_danmu_to_file(danmu):

首先寫入文本日志：

with open(LOG_FILE, 'a', encoding='utf-8') as f:
    f.write(danmu['formatted'] + '\n\n')

再寫入 JSON 數(shù)據(jù)：

json.dump(danmu_with_timestamp, f, ensure_ascii=False)

并增加保存時間：

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"
}

這樣既可以用于日志查看，也可以直接用于數(shù)據(jù)分析。

八、程序啟動與關(guān)閉

主程序入口：

if __name__ == "__main__":

程序啟動時會記錄日志：

==================================================
程序啟動于: 2026-03-09 20:30:01
==================================================

程序結(jié)束時也會記錄退出時間。

這一步設(shè)計的好處是：可以知道采集時長，也可以追蹤異常退出。

九、彈幕采集程序的設(shè)計心得

在設(shè)計這個彈幕采集程序時，我有幾個比較深的體會：

第一，WebSocket非常適合實時數(shù)據(jù)采集。相比傳統(tǒng)HTTP輪詢方式，WebSocket能夠持續(xù)保持連接，從而實時接收彈幕消息，大幅降低延遲。

第二，異步編程是處理高頻消息流的關(guān)鍵。使用 asyncio 可以避免阻塞，提高程序整體效率。

第三，數(shù)據(jù)存儲需要兼顧實時與分析需求。因此同時保存為日志文件和 JSON 文件，這樣既能實時查看，也方便后續(xù)做數(shù)據(jù)分析。

第四，緩存隊列設(shè)計可以提高系統(tǒng)穩(wěn)定性。通過 deque 保存最近彈幕，可以避免高頻消息導(dǎo)致系統(tǒng)壓力過大。

十、總結(jié)

通過這套程序，我們實現(xiàn)了一套完整的 直播彈幕實時采集系統(tǒng)，具備以下特點：

（1）WebSocket實時采集

（2）asyncio異步處理

（3）彈幕隊列緩存

（4）自動日志記錄

（5）JSON結(jié)構(gòu)化數(shù)據(jù)保存

（6）程序運行狀態(tài)記錄

這套程序不僅可以用于直播彈幕采集，還可以擴展到實時輿情監(jiān)測、彈幕情感分析、用戶互動研究、直播熱點識別。

后續(xù)如果需要，還可以進一步加入數(shù)據(jù)庫存儲（MySQL / MongoDB）、彈幕情感分析、實時數(shù)據(jù)可視化、彈幕關(guān)鍵詞統(tǒng)計。

從整體實現(xiàn)來看，這套彈幕采集程序雖然代碼量不算很大，但在設(shè)計上我更關(guān)注的是穩(wěn)定性和可擴展性。通過 WebSocket 實現(xiàn)實時數(shù)據(jù)接收，再結(jié)合 asyncio 異步處理，可以比較輕松地應(yīng)對高頻彈幕消息；同時利用 deque 做緩存隊列、日志系統(tǒng)做運行記錄、JSON 做結(jié)構(gòu)化數(shù)據(jù)存儲，使得整個數(shù)據(jù)流從“接收 → 解析 → 緩存 → 落盤”形成一個完整閉環(huán)。這樣的設(shè)計不僅方便后續(xù)做彈幕文本分析、情感分析或關(guān)鍵詞統(tǒng)計，也為進一步接入數(shù)據(jù)庫、實時可視化或數(shù)據(jù)流處理打下了基礎(chǔ)。

以上就是Python基于WebSocket實現(xiàn)直播彈幕數(shù)據(jù)采集的詳細內(nèi)容，更多關(guān)于Python WebSocket彈幕數(shù)據(jù)采集的資料請關(guān)注腳本之家其它相關(guān)文章！

最新評論