秒级行情推送系统实战：从触发、采集到入库的端到端架构

原创

jackcode

发布于 2025-09-28 11:30:23

15200

代码可运行

文章被收录于专栏：爬虫资料爬虫资料

运行总次数：0

代码可运行

爬虫代理

在金融业务中，实时行情推送 是关键能力：用户希望在秒级内看到股票、数字货币或大宗商品的价格变化。传统的批量爬取往往延迟较高，无法满足交易和风控的需要。本文将结合 行情业务场景，搭建一个 秒级数据更新与推送系统，并演示端到端架构。

一、系统架构图

         ┌────────────────┐
         │   触发层 (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)  # 每秒执行一次

代码要点：