微服务化采集平台：可扩展性与容错机制

在资本市场博弈中，信息永远是先手优势。财经资讯，尤其是突发事件、政策信号、个股动态，往往在最初10分钟内的舆论发酵期影响最大。能否及时捕捉这些“情绪燃点”，决定了一个投研系统的数据基础是否够强。

然而，在实际调研中我们发现：如新浪财经这类站点虽然内容丰富、更新频繁，但其信息组织方式并非为程序分析而优化。新闻分布在多个频道，结构经常变动，突发类内容散落无序，人工整理效率极低。

因此，本文将以一个真实项目为例，从财经视角出发构建一个微服务化采集平台，实现：

• 精准提取要闻、突发、证券资讯三类数据；
• 自动分类与统计分析；
• 模块解耦、弹性伸缩；
• 提供数据支撑能力，为后续NLP、风控建模等场景服务。

一、财经视角下的需求与痛点

在观察了若干财经内容站点后，我们归纳出几类共性问题：

如果没有一套结构化采集+智能归类+容错机制并存的架构支撑，单靠传统工具采集财经数据，将永远慢市场一步。

二、平台设计总览（系统技术关系图）

以下是平台的微服务模块关系图，每一部分均可独立部署与维护，真正实现解耦与扩展性：

• 调度器：下发任务，控制执行顺序与频率；
• 采集节点：分别抓取对应频道内容；
• 解析清洗服务：统一提取标题、正文、时间；
• 统计分析服务：实现关键词归类、每日数据统计；
• 容错机制：对失败请求进行重试、上报；
• 代理服务：对接第三方代理，如亿牛云，解决封锁问题。

三、要闻采集模块示例

import requests
from lxml import etree
import random
import time

# --- 代理配置（参考亿牛云代理 www.16yun.cn） ---
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}"
}

# --- 随机User-Agent池 ---
user_agents = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64)...",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)..."
]

# --- 请求头构建 ---
headers = {
    "User-Agent": random.choice(user_agents),
    "Cookie": "U_TRS1=xxxxx; U_TRS2=xxxxx;"
}

# --- 抓取新浪首页要闻链接 ---
def fetch_top_news():
    url = "https://finance.sina.com.cn/"
    try:
        res = requests.get(url, headers=headers, proxies=proxies, timeout=10)
        res.encoding = 'utf-8'
        tree = etree.HTML(res.text)
        news_items = tree.xpath('//div[@id="blk_yw_01"]/ul/li/a')
        result = []
        for item in news_items:
            title = item.xpath("text()")[0]
            link = item.xpath("@href")[0]
            result.append((title.strip(), link.strip()))
        return result
    except Exception as e:
        print("⚠️ 采集失败：", e)
        return []

四、正文解析模块（提取标题+时间+正文）

def fetch_news_detail(link):
    try:
        res = requests.get(link, headers=headers, proxies=proxies, timeout=10)
        tree = etree.HTML(res.text)
        title = tree.xpath('//h1/text()')[0]
        content = "\n".join(tree.xpath('//div[@id="artibody"]//p/text()'))
        time_str = tree.xpath('//span[@class="date"]/text()')
        return {
            "title": title.strip(),
            "content": content.strip(),
            "time": time_str[0] if time_str else "未知时间"
        }
    except Exception as e:
        return {"error": str(e)}

五、财经内容分类与统计模块

from collections import defaultdict
import json

def classify(data_list):
    stats = defaultdict(int)
    classified = defaultdict(list)

    keywords = {
        "要闻": ["GDP", "政策", "A股", "央行"],
        "突发": ["突发", "事故", "爆雷", "暴跌"],
        "证券": ["股价", "涨停", "跌停", "交易"]
    }

    for item in data_list:
        matched = False
        for tag, kwlist in keywords.items():
            if any(kw in item["title"] for kw in kwlist):
                stats[tag] += 1
                classified[tag].append(item)
                matched = True
                break
        if not matched:
            stats["未分类"] += 1
            classified["未分类"].append(item)

    print("分类统计结果：")
    print(json.dumps(stats, ensure_ascii=False, indent=2))
    return classified

六、异常捕捉与请求重试机制

import functools
import time

def retry(max_retry=3, wait=2):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for i in range(max_retry):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    print(f"第{i+1}次重试失败：{e}")
                    time.sleep(wait)
            return {"error": "全部重试失败"}
        return wrapper
    return decorator

@retry(max_retry=3)
def safe_request(url):
    return requests.get(url, headers=headers, proxies=proxies, timeout=8)

七、平台价值与进阶方向

通过以上模块，我们实现了一个具备高可用性、可维护性与财经领域特化能力的分布式采集平台。它不仅可以支撑日常的信息获取，更为舆情检测、投研辅助系统提供了可持续数据能力。