本文目录导读：
- 基础工具安装
- 方法一：使用websocket-client库（同步）
- 方法二：使用websockets库（异步）
- 方法三：抓取加密交易所WebSocket示例
- 方法四：使用Selenium模拟浏览器WebSocket
- 高级特性：自动重连与数据持久化
- 注意事项
我来详细说明如何使用Python爬虫抓取WebSocket数据。
基础工具安装
# 安装websocket客户端库 pip install websocket-client pip install websockets # 异步版本 # 可选工具 pip install aiohttp # 异步HTTP客户端，支持WebSocket pip install ws4py # 另一种WebSocket客户端
使用websocket-client库（同步）
import websocket
import json
import threading
def on_message(ws, message):
"""收到消息时的回调"""
data = json.loads(message)
print(f"收到数据: {data}")
# 这里处理你的数据，如保存到文件或数据库
def on_error(ws, error):
"""错误处理"""
print(f"错误: {error}")
def on_close(ws, close_status_code, close_msg):
"""连接关闭处理"""
print(f"连接关闭: {close_status_code} - {close_msg}")
def on_open(ws):
"""连接成功建立"""
print("WebSocket连接已建立")
# 发送订阅消息
subscribe_msg = {
"event": "subscribe",
"channel": "ticker",
"symbol": "BTC/USDT"
}
ws.send(json.dumps(subscribe_msg))
def connect_websocket():
"""建立WebSocket连接"""
# 禁用SSL验证（如果需要）
websocket.enableTrace(False)
# 连接WebSocket
ws = websocket.WebSocketApp(
"wss://your-websocket-url.com/ws", # 替换为实际URL
on_open=on_open,
on_message=on_message,
on_error=on_error,
on_close=on_close
)
# 运行连接（会阻塞）
ws.run_forever()
# 启动连接
connect_websocket()
使用websockets库（异步）
import asyncio
import websockets
import json
async def subscribe_market_data():
"""订阅市场数据"""
uri = "wss://your-websocket-url.com/ws" # 替换为实际URL
async with websockets.connect(uri) as websocket:
# 发送订阅请求
subscribe_msg = {
"event": "subscribe",
"channel": "ticker",
"symbol": "BTC/USDT"
}
await websocket.send(json.dumps(subscribe_msg))
# 持续接收数据
while True:
try:
response = await websocket.recv()
data = json.loads(response)
print(f"收到数据: {data}")
# 数据处理逻辑
# save_to_database(data)
except websockets.exceptions.ConnectionClosed:
print("连接已关闭，尝试重连...")
break
except Exception as e:
print(f"错误: {e}")
break
async def main():
"""主函数"""
# 创建重连机制
while True:
try:
await subscribe_market_data()
except Exception as e:
print(f"连接失败: {e}")
await asyncio.sleep(5) # 等待5秒重试
# 运行
asyncio.run(main())
抓取加密交易所WebSocket示例
import websocket
import json
import threading
import time
from datetime import datetime
class CryptoWebSocket:
"""加密货币WebSocket抓取器"""
def __init__(self, url, channels=None):
self.url = url
self.channels = channels or []
self.ws = None
self.running = False
self.reconnect_count = 0
def on_message(self, ws, message):
"""处理接收到的消息"""
try:
data = json.loads(message)
# 根据数据类型处理
if 'channel' in data:
channel = data['channel']
if channel == 'ticker':
self.handle_ticker(data)
elif channel == 'trade':
self.handle_trade(data)
elif channel == 'orderbook':
self.handle_orderbook(data)
except json.JSONDecodeError:
print(f"无法解析的JSON数据: {message[:100]}")
def handle_ticker(self, data):
"""处理ticker数据"""
ticker_data = {
'timestamp': datetime.now().isoformat(),
'symbol': data.get('symbol'),
'price': data.get('price'),
'volume': data.get('volume'),
'change': data.get('change')
}
print(f"Ticker: {ticker_data}")
# 保存到文件或数据库
def handle_trade(self, data):
"""处理交易数据"""
print(f"Trade: {json.dumps(data, indent=2)}")
def handle_orderbook(self, data):
"""处理订单簿数据"""
print(f"Orderbook update received")
def on_error(self, ws, error):
"""处理错误"""
print(f"WebSocket错误: {error}")
def on_close(self, ws, close_status_code, close_msg):
"""处理连接关闭"""
print(f"WebSocket连接关闭: {close_status_code}")
if self.running:
self.reconnect()
def on_open(self, ws):
"""处理连接建立"""
print("WebSocket连接已建立")
self.reconnect_count = 0
# 发送订阅消息
for channel in self.channels:
subscribe_msg = {
"event": "subscribe",
"channel": channel
}
ws.send(json.dumps(subscribe_msg))
print(f"已订阅: {channel}")
def connect(self):
"""建立连接"""
websocket.enableTrace(False)
self.ws = websocket.WebSocketApp(
self.url,
on_open=self.on_open,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close
)
self.running = True
# 在另一个线程中运行
wst = threading.Thread(target=self.ws.run_forever)
wst.daemon = True
wst.start()
def reconnect(self):
"""重连机制"""
if self.reconnect_count < 10: # 最多重连10次
wait_time = min(30, 2 ** self.reconnect_count)
print(f"将在 {wait_time} 秒后重连...")
time.sleep(wait_time)
self.reconnect_count += 1
self.connect()
def disconnect(self):
"""断开连接"""
self.running = False
if self.ws:
self.ws.close()
def send_message(self, message):
"""发送消息"""
if self.ws and self.running:
self.ws.send(json.dumps(message))
# 使用示例
def main():
# 配置WebSocket连接
ws_client = CryptoWebSocket(
url="wss://stream.binance.com:9443/ws",
channels=["btcusdt@ticker", "ethusdt@ticker"]
)
# 启动连接
ws_client.connect()
try:
# 保持主线程运行
while True:
time.sleep(1)
except KeyboardInterrupt:
print("正在关闭连接...")
ws_client.disconnect()
if __name__ == "__main__":
main()
使用Selenium模拟浏览器WebSocket
from selenium import webdriver
from selenium.webdriver.common.by import By
import json
import time
def capture_websocket_via_selenium():
"""通过Selenium捕获WebSocket通信"""
options = webdriver.ChromeOptions()
options.add_argument('--disable-gpu')
options.add_argument('--headless') # 无头模式
# 启用性能日志
options.set_capability('goog:loggingPrefs', {'performance': 'ALL'})
driver = webdriver.Chrome(options=options)
try:
# 访问目标网站
driver.get("https://example.com") # 替换为实际URL
# 等待页面加载
time.sleep(5)
# 获取性能日志
logs = driver.get_log('performance')
websocket_messages = []
for entry in logs:
log_data = json.loads(entry['message'])
message = log_data.get('message', {})
# 过滤WebSocket消息
if message.get('method') == 'Network.webSocketFrameReceived':
params = message.get('params', {})
response = params.get('response', {})
payload = response.get('payloadData', '')
try:
data = json.loads(payload)
websocket_messages.append(data)
print(f"WebSocket消息: {data}")
except:
pass
return websocket_messages
finally:
driver.quit()
高级特性：自动重连与数据持久化
import asyncio
import websockets
import json
import sqlite3
from datetime import datetime
import logging
logging.basicConfig(level=logging.INFO)
class WebSocketCrawler:
"""高级WebSocket爬虫"""
def __init__(self, url, db_path='websocket_data.db'):
self.url = url
self.db_path = db_path
self.setup_database()
def setup_database(self):
"""初始化数据库"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# 创建数据表
cursor.execute('''
CREATE TABLE IF NOT EXISTS websocket_data (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT,
channel TEXT,
data TEXT,
received_at TEXT
)
''')
conn.commit()
conn.close()
def save_to_database(self, channel, data):
"""保存数据到数据库"""
try:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
INSERT INTO websocket_data (timestamp, channel, data, received_at)
VALUES (?, ?, ?, ?)
''', (
datetime.now().isoformat(),
channel,
json.dumps(data),
datetime.now().isoformat()
))
conn.commit()
conn.close()
return True
except Exception as e:
logging.error(f"保存数据失败: {e}")
return False
async def connect_with_retry(self, max_retries=5):
"""带重试的连接"""
for attempt in range(max_retries):
try:
async with websockets.connect(
self.url,
ping_interval=30,
ping_timeout=10
) as websocket:
logging.info(f"连接成功 (尝试 {attempt + 1})")
# 订阅数据
await websocket.send(json.dumps({
"event": "subscribe",
"channel": "ticker"
}))
# 持续接收数据
async for message in websocket:
try:
data = json.loads(message)
channel = data.get('channel', 'unknown')
# 保存到数据库
self.save_to_database(channel, data)
# 可选：发送到消息队列
# await self.send_to_queue(data)
except json.JSONDecodeError:
continue
except websockets.exceptions.ConnectionClosed:
logging.warning(f"连接断开，准备重试 ({attempt + 1}/{max_retries})")
await asyncio.sleep(2 ** attempt) # 指数退避
except Exception as e:
logging.error(f"连接错误: {e}")
await asyncio.sleep(5)
logging.error("达到最大重试次数，放弃连接")
# 使用示例
async def main():
crawler = WebSocketCrawler("wss://your-websocket-url.com/ws")
await crawler.connect_with_retry()
if __name__ == "__main__":
asyncio.run(main())
注意事项
- 法律合规：确保抓取行为符合网站服务条款和当地法律法规
- 频率限制：遵守WebSocket的速率限制，避免过度请求
- 数据解析：不同平台的数据格式可能不同，需要适配解析
- 错误处理：实现完善的错误处理和重连机制
- 资源管理：及时关闭连接，释放资源
- 安全性：避免在代码中硬编码敏感信息
选择哪种方法取决于你的具体需求：同步库适合简单场景，异步库适合高并发，Selenium适合抓取动态加载的WebSocket数据。
