当前位置：首页 > news >正文

优化分布式采集的数据同步：一致性、去重与冲突解决的那些坑与招

news 2025/10/14 6:23:24

写采集的人都知道，真正让人头疼的，往往不是抓不下来，而是抓下来的数据不对劲。
我曾经被这个问题折磨到怀疑人生。直到有一天，我决定好好把“同步”这件事解决干净。

一、那次混乱的分布式采集任务

几年前，我接了个房地产数据采集项目。任务看起来很普通：
每天从几十个房产网站抓取新房源，做价格走势分析。

最初一切顺利，直到我们把采集方案扩展到十几台服务器那天，数据库开始“闹鬼”。

一套房源被存了五次；有些价格明明变了，但我们那边还是旧的；甚至还有两台节点同时写入同一条数据，结果字段被覆盖。
我花了好几天对日志、对表、对时间线，才意识到问题根本不在采集，而是在数据同步这一环。

二、线索：混乱的根源不止一个

那阵子我几乎泡在日志里，每一条异常都追到源头。
最后发现有三个主要问题：

首先是写入冲突。不同节点在同一时间采到同一条房源，互相覆盖。
其次是旧数据反超新数据。有些节点延迟太大，旧内容却被当成“最新”。
最后是去重困难。房源URL、ID、标题都不稳定，没法唯一识别。

这三件事加起来，就像三只不听话的猫：一个改字段、一个乱更新、一个重复喂食。
最终的结果是，系统跑得飞快，数据却乱成一锅粥。

三、破局：我总结出的“三板斧”

我没有推倒重来，而是硬着头皮在原有架构上做了系列优化。
最后靠三板斧稳住了局面：一致性、去重、冲突解决。

第一板斧：一致性——用时间戳和哈希说话

我给每条采集到的数据都加了两个字段：一个是时间戳 update_ts，记录采集时间；另一个是 hash_sig，用来表示页面内容的哈希值。

逻辑非常简单：当数据入库时，如果新数据的时间更新、内容也不同，就覆盖旧的；否则跳过。
这样一来，即使多个节点重复采集，也不会导致数据混乱。
这个设计的核心思想是“幂等性”，也就是多次执行结果保持一致。

第二板斧：去重——URL归一化 + Redis布隆过滤器

不同节点抓到的URL往往不一样。
例如：

https://example.com/house?id=123
https://example.com/house/123

我先做了URL归一化：去掉参数、补上路径。
然后用Redis布隆过滤器判断是否采过。
这一改，重复采集的比率直接从两位数降到个位数。

示例代码（爬虫代理配置）

import hashlib
from urllib.parse import urlparse
import redis
import time
import requests# ======== 亿牛云爬虫代理配置 www.16yun.cn========
proxy_host = "proxy.16yun.cn"
proxy_port = "12345"
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}"
}# ======== Redis连接，用于分布式同步 ========
r = redis.Redis(host="127.0.0.1", port=6379, db=0)def normalize_url(url):"""URL归一化"""parsed = urlparse(url)return f"{parsed.scheme}://{parsed.netloc}{parsed.path}"def url_to_hash(url):"""生成URL哈希"""return hashlib.md5(normalize_url(url).encode()).hexdigest()def is_new_url(url):"""用Redis布隆过滤器判断是否已采集"""h = url_to_hash(url)key = f"bloom:url:{h[:2]}"return r.setnx(key, h)def crawl(url):"""简单的采集逻辑"""if not is_new_url(url):print(f"[跳过重复] {url}")returntry:resp = requests.get(url, proxies=proxies, timeout=10)print(f"[成功] {url}, 内容长度={len(resp.text)}")data = {"url": url,"update_ts": int(time.time()),"hash_sig": hashlib.md5(resp.text.encode()).hexdigest()}# 数据库写入逻辑略except Exception as e:print(f"[失败] {url}, 原因: {e}")

第三板斧：冲突解决——加锁，而不是“硬拼”

节点多了之后，偶尔会出现两个节点同时写同一条记录。
最开始我用延迟重试，但依旧偶尔撞车。
后来改成在Redis中加分布式锁，谁先拿到锁，谁写；写完再释放。

def write_with_lock(url, data):"""防止多个节点同时写入"""lock_key = f"lock:{url_to_hash(url)}"if not r.set(lock_key, 1, nx=True, ex=5):  # 尝试加锁print(f"[锁被占用] {url}")returntry:print(f"[写入中] {url}")# 模拟数据库操作……time.sleep(1)finally:r.delete(lock_key)