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

一、系统架构图

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

代码要点：

1. 采集层 使用 交易所 API 获取实时价格。
2. 代理IP（爬虫代理） 确保高频访问不限制。
3. 入库层 存储行情，用于历史分析。
4. 推送层 模拟实时推送，实际可用 WebSocket 广播。

四、总结

在行情业务中，秒级更新系统 的核心目标是：

• 触发层：秒级驱动行情采集。
• 采集层：API + 代理IP 防止封禁。
• 缓冲层：保障高并发稳定。
• 入库层：保存历史+缓存实时数据。
• 推送层：保证用户端 秒级可见行情。

这种端到端架构，既能满足 高并发，又能做到 低延迟，是 行情推送、金融交易风控、实时监控 等场景的理想方案。