在金融业务中，实时行情推送 是关键能力：用户希望在 秒级 内看到股票、数字货币或大宗商品的价格变化。传统的批量爬取往往延迟较高，无法满足交易和风控的需要。本文将结合 行情业务场景，搭建一个 秒级数据更新与推送系统，并演示端到端架构。
一、系统架构图
┌────────────────┐
│ 触发层 (Trigger)│ ← 定时调度/交易所推送事件
└──────┬─────────┘
│
┌──────▼─────────┐
│ 采集层 (Crawler)│ ← 请求交易所行情API / 爬虫采集
│ (代理IP + UA )│
└──────┬─────────┘
│
┌──────▼─────────┐
│ 缓冲层 (Buffer) │ ← Kafka / Redis Stream
└──────┬─────────┘
│
┌──────▼─────────┐
│ 入库层 (DB) │ ← MySQL / Redis 缓存
└──────┬─────────┘
│
┌──────▼─────────┐
│ 推送层 (Push) │ ← WebSocket / 消息队列 → 用户终端
└────────────────┘
与通用爬虫不同，行情业务的关键在于 低延迟 + 高并发，所以在采集后，数据会先进入缓冲层，再统一推送。
二、模块功能介绍
1. 触发层（Trigger）
- 来源：定时任务（如每秒触发一次）、交易所 API WebSocket 事件。
- 功能：保证行情数据在 秒级 被拉取。
2. 采集层（Crawler）
- 获取 股票/币价 最新行情。
- 使用 代理IP（爬虫代理） 防止 API 请求被限流或封禁。
3. 缓冲层（Buffer）
- 使用 Kafka/Redis Stream 暂存行情，防止入库时拥塞。
- 支持多消费者：一个消费者负责入库，另一个负责推送用户。
4. 入库层（DB）
- 写入 MySQL（历史行情存储）、Redis（实时缓存）。
- 支持前端秒级查询和分析。
5. 推送层（Push）
- 将最新行情 秒级推送 给用户。
- 技术手段：WebSocket / 消息队列（RabbitMQ）。
三、关键代码详解（行情采集与推送）
下面以 Python + requests + MySQL 为例，采集实时行情（模拟从交易所接口获取 BTC/USDT 最新价格），并写入数据库。
import requests
import pymysql
import time
# ====== 代理配置（亿牛云爬虫代理示例） ======
proxy_host = "proxy.16yun.cn" # 代理域名
proxy_port = "3100" # 代理端口
proxy_user = "16YUN" # 用户名
proxy_pass = "16IP" # 密码
proxies = {
"http": f"http://{proxy_user}:{proxy_pass}@{proxy_host}:{proxy_port}",
"https": f"http://{proxy_user}:{proxy_pass}@{proxy_host}:{proxy_port}"
}
# ====== 数据库配置 ======
db = pymysql.connect(
host="localhost",
user="root",
password="123456",
database="market_db",
charset="utf8mb4"
)
cursor = db.cursor()
# ====== 采集函数（模拟获取行情API） ======
def fetch_price(symbol="BTCUSDT"):
url = f"https://api.binance.com/api/v3/ticker/price?symbol={symbol}"
try:
resp = requests.get(url, proxies=proxies, timeout=5)
data = resp.json()
price = float(data["price"])
return price
except Exception as e:
print("行情获取失败：", e)
return None
# ====== 数据入库函数 ======
def save_to_db(symbol, price):
sql = "INSERT INTO market_price (symbol, price, create_time) VALUES (%s, %s, NOW())"
try:
cursor.execute(sql, (symbol, price))
db.commit()
print(f"[入库成功] {symbol} 最新价格: {price}")
except Exception as e:
print("写库失败：", e)
db.rollback()
# ====== 推送层（模拟，实际可用 WebSocket） ======
def push_to_client(symbol, price):
print(f"[推送到客户端] {symbol} 最新价格 {price}")
# ====== 秒级调度（触发层） ======
symbols = ["BTCUSDT", "ETHUSDT"]
while True:
for sym in symbols:
price = fetch_price(sym)
if price:
save_to_db(sym, price)
push_to_client(sym, price)
time.sleep(1) # 每秒执行一次
代码要点：
- 采集层 使用 交易所 API 获取实时价格。
- 代理IP（爬虫代理） 确保高频访问不限制。
- 入库层 存储行情，用于历史分析。
- 推送层 模拟实时推送，实际可用 WebSocket 广播。
四、总结
在行情业务中，秒级更新系统 的核心目标是：
- 触发层：秒级驱动行情采集。
- 采集层：API + 代理IP 防止封禁。
- 缓冲层：保障高并发稳定。
- 入库层：保存历史+缓存实时数据。
- 推送层：保证用户端 秒级可见行情。
这种端到端架构，既能满足 高并发，又能做到 低延迟，是 行情推送、金融交易风控、实时监控 等场景的理想方案。
