如果你做过分布式采集，一定遇到过这种场景：

+ 任务量一上来，节点越加越多
+ URL 重复抓、反复抓、疯狂抓
+ 明明“成功抓取”的日志写满屏，结果入库时发现有一堆漏网之鱼
+ 代理挂了、节点奔溃了、队列卡住了……
+ 最后复盘时，只剩一句“怎么又丢了？”

听上去很熟悉吧？ 
这就是分布式采集中永恒的痛点：**一致性问题**。

本文就想把这件事说清楚：从痛点、到原理、到工程化方案，再到可运行的示例代码。你会看到一个完整的闭环系统，告诉你如何让每天早上 8 点去抓“中国政府采购网”公告，既不重复，也不错漏。

是的，我们还会顺手加上代理、定时任务、Redis 原子去重、Stream 可靠队列、消费者组等一整套工业级思路。

# 一、为什么分布式采集很容易“不一致”？
先别急着谈架构，我们从几个真实发生过的“坑”说起。

### 1. 单机去重：简单好用，但一扩容就碎成渣
本地 SQLite、文本哈希表、单机字典……在一台机器上挺香。但当你挂了十几台 worker 时，每台都各玩各的，没有一个共享状态，重复抓取根本挡不住。

### 2. URL 哈希去重：能挡部分重复，却挡不住竞态
常见的策略是：

+ 先查数据库有没有
+ 没有就插入 
这个“先查后写”的两步操作，只要并发一高，就可以出现著名的“查的时候没有，但写的时候重复了”问题。

### 3. 消息队列：天然不是数据库
很多团队把 RabbitMQ / Kafka 当存储用，实际上它们并不是为“可靠最终入库”而设计的。只要消费者在「写入数据库之前」崩了，你就会出现 “消息被消费，但数据没入库” 的经典丢失问题。

### 4. 高并发 + 代理失效：失败重试导致“雪崩式重复”
代理抖动是常态。如果你不做合理的“失败重试 + 幂等判断”，每个节点都会以为“刚刚网络失败，所以我要再来一次”。

换句话说： 
**重复抓取是工程缺陷，丢失数据是流程缺陷。**

要解决它们，我们需要一个真正分布式的一致性方案，而不仅是“分布式的采集脚本”。

---

# 二、一个靠谱的架构应该长什么样？（文字版架构图 + 模块拆解）
我们把整个系统拆成五个部分：

**定时触发 → 抓取列表 → 原子去重 → 任务流转 → 消费持久化**

下面这个是“文字版架构图”，你可以直接脑内成像：

```plain
定时任务（每天8点）
 ↓
抓取模块（使用爬虫代理）
 ↓
提取公告ID（核心唯一标识）
 ↓
Redis 原子去重（SET + Lua）
 ↓
Redis Stream（可靠队列）
 ↓
多个 Worker（消费者组）
 ↓
数据库（带唯一键的幂等写入）
```

再来看看每个模块的职责——更像人说话的版本：

### （1）定时器：每天早上 8 点叫醒整个系统
APScheduler 负责叫醒采集，它就像闹钟一样准时，只不过是给机器叫。

### （2）抓取层：翻页、解析、抽公告
这部分要解决的不是“能不能抓”，而是“能抓多久”。 
这就是为什么代理（这里用爬虫代理）和 headers、cookies 这些东西必须提前调好。

### （3）去重层：灵魂所在，不然全系统白忙
我们用 Redis SET 做全局唯一标记，并且用 Lua 脚本保证“判断存在 + 加入集合 + 写队列”是一个原子操作。

这是整个系统的关键点。 
**没有原子性，你的去重逻辑永远会被竞态打爆。**

### （4）队列层：不只是队列，而是“可靠投递”
Redis Stream + Consumer Group 是非常适合采集分布式任务流转的组件：

+ 任务会被持久化
+ 消费者挂了不会丢
+ pending 队列能帮助检查未完成任务
+ 可以多消费者并行，提高吞吐

### （5）入库层：幂等是所有悲剧的尽头
数据库必须加唯一索引，并使用：

+ PostgreSQL：`ON CONFLICT DO NOTHING`
+ MySQL：`INSERT … ON DUPLICATE KEY UPDATE`

否则你前面做的所有去重努力，到了数据库这一层都没意义。

---

# 三、性能差异与行业实践：为什么“原子去重 + Stream”是主流方案？
我们测试过三种常见系统：

| 模式 | 分布式竞争情况 | 重复率 | 丢失率 | 性能 |
| --- | --- | --- | --- | --- |
| 简单数据库去重 | 高 | 中高 | 中 | 慢 |
| Redis SET 去重（无 Lua） | 中 | 中 | 低 | 快 |
| Redis SET + Lua 原子 + Stream | 低 | 极低 | 极低 | 快/可水平扩展 |


两点关键差异特别值得说：

### **1. Redis 原子脚本不是玄学，是工程师的救命稻草**
竞态之所以可怕，是因为它“发生的频率很小，但一旦发生就毁数据”。 
Lua 脚本让“判断 + 加入 + 发队列”成为单指令事务，这一步是必须的。

### **2. Stream 的 pending 队列可以救你一命**
消费者挂了？ 
崩了？ 
断网了？ 
进程被 Kill -9 了？

你不需要猜。 
所有“已读未确认”的消息都还躺在 pending 队列里，只要把它“claim”回来就能继续处理。

这就是为什么越来越多行业的增量采集项目——政务监测、电商监控、舆情系统、财经公告、采集 SaaS 平台——都采用「Redis Stream + 去重 SET」作为底层任务流转架构。

---

# 四、技术演化：采集一致性的进化史
我们把常见技术方案做了一棵演化树（文字版）：

```plain
单机文件去重
 ↓（扩容需求）
SQLite / 本地DB去重
 ↓（分布式冲突）
数据库唯一索引去重
 ↓（性能瓶颈 + 竞态）
Redis SET 去重
 ↓（一致性不足）
Redis SET + Lua 原子脚本
 ↓（任务丢失）
Redis Stream + Consumer Group
 ↓（大规模并发）
SET/Bloom + Lua + Stream + 幂等DB（当前主流）
```

一句话总结： 
**所有进化都是为了减少“重复”与“丢失”。**

# 五、示例代码
下面是可运行的参考实现，分为 “爬取生产者” 和 “入库消费者”。

## 1）producer.py —— 定时抓取 + 原子去重 + 写入 Stream
```python
# producer.py
"""
每天 08:00 抓取全国政府采购网公告（示例）
流程：抓列表 -> 提取公告ID -> Redis 原子去重 -> 写 Redis Stream
"""

import requests
from bs4 import BeautifulSoup
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
import redis
import pytz
import json

# ======== 16YUN爬虫代理 ========
PROXY_HOST = "proxy.16yun.cn"
PROXY_PORT = 12345
PROXY_USER = "your_proxy_username"
PROXY_PASS = "your_proxy_password"

PROXY_URL = f"http://{PROXY_USER}:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}"

# ======== Redis 连接 ========
r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

REDIS_SEEN_SET = "govproc:seen" # 全局去重集合
REDIS_STREAM = "govproc:stream" # 任务队列（可靠）

# Lua 原子检查：若不存在则加入去重集合并写入 Stream
LUA_CHECK_ADD_STREAM = """
local seen_key = KEYS[1]
local stream_key = KEYS[2]
local id = ARGV[1]
local payload = ARGV[2]

local exists = redis.call('SISMEMBER', seen_key, id)
if exists == 1 then
 return 0
end

redis.call('SADD', seen_key, id)
local msg_id = redis.call('XADD', stream_key, '*', 'payload', payload)
return msg_id
"""
lua_check_add = r.register_script(LUA_CHECK_ADD_STREAM)

# ======== 请求会话（含代理） ========
session = requests.Session()
session.proxies = {"http": PROXY_URL, "https": PROXY_URL}
session.headers.update({
 "User-Agent": "Mozilla/5.0 (compatible; GovProcBot/1.0)",
})
session.timeout = 20

TARGET_LIST_URL = "https://www.ccgp.gov.cn/cggg/zbgg/" # 示例

def fetch_list():
 try:
 resp = session.get(TARGET_LIST_URL)
 resp.raise_for_status()
 return BeautifulSoup(resp.text, "html.parser")
 except Exception as e:
 print("请求失败：", e)
 return None

def parse_list(soup):
 results = []
 # 以下选择器仅为示例，请按真实页面结构调整
 for li in soup.select("ul.news-list li"):
 a = li.find("a")
 if not a:
 continue
 url = a.get("href")
 title = a.get_text(strip=True)
 ann_id = url.split("/")[-1] # 示例做 ID
 results.append((ann_id, title, url))
 return results

def handle_list():
 print("开始抓取列表页...")
 soup = fetch_list()
 if not soup:
 return

 announcements = parse_list(soup)
 for ann_id, title, url in announcements:
 payload = json.dumps({"id": ann_id, "title": title, "url": url})

 msg_id = lua_check_add(keys=[REDIS_SEEN_SET, REDIS_STREAM],
 args=[ann_id, payload])

 if msg_id == 0:
 print(f"重复跳过：{ann_id}")
 else:
 print(f"加入任务：{ann_id} -> {msg_id}")

def main():
 scheduler = BlockingScheduler(timezone=pytz.timezone("Asia/Shanghai"))
 scheduler.add_job(handle_list, CronTrigger(hour=8, minute=0))
 print("任务已启动（每天 08:00 执行）")
 scheduler.start()

if __name__ == "__main__":
 main()
```

## 2）consumer.py —— 消费 Stream + 幂等入库
```python
# consumer.py
"""
从 Redis Stream 消费任务，并写入数据库（示例使用 PostgreSQL）
"""

import redis
import json
from sqlalchemy import create_engine, text

# ======== Redis 配置 ========
r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
STREAM = "govproc:stream"
GROUP = "gov-workers"
CONSUMER = "worker-1"

# 创建消费组（如果不存在）
try:
 r.xgroup_create(STREAM, GROUP, id="0", mkstream=True)
except Exception:
 pass # 存在则忽略

# ======== 数据库连接 ========
engine = create_engine(
 "postgresql://username:password@localhost:5432/mydb"
)

# 表结构需要带唯一索引（例如 id 作为唯一键）
INSERT_SQL = """
INSERT INTO gov_procure (id, title, url)
VALUES (:id, :title, :url)
ON CONFLICT (id) DO NOTHING
"""

def process_one(msg_id, payload):
 """处理单条消息"""
 data = json.loads(payload)
 with engine.begin() as conn:
 conn.execute(text(INSERT_SQL), {
 "id": data["id"],
 "title": data["title"],
 "url": data["url"],
 })

def main():
 print("消费者启动...")
 while True:
 msgs = r.xreadgroup(
 GROUP, CONSUMER,
 streams={STREAM: ">"}, # 只读未处理的新消息
 count=10, block=5000
 )

 if not msgs:
 continue

 for stream, entries in msgs:
 for msg_id, fields in entries:
 payload = fields["payload"]
 try:
 process_one(msg_id, payload)
 r.xack(STREAM, GROUP, msg_id)
 print("完成：", msg_id)
 except Exception as e:
 print("入库失败：", msg_id, e)

if __name__ == "__main__":
 main()
```

# 六、总结：真正稳定的系统靠“设计”
分布式采集之所以容易踩坑，不是因为写采集难，而是整个系统链路足够长、足够多点断层。

你现在看到的这套方案，其实就是行业普遍收敛出来的最佳路线：

+ **SET 做全局唯一 + Lua 做原子性**
+ **Stream 做可靠队列 + Consumer Group 做水平扩展**
+ **数据库使用幂等写入**
+ **代理管理、重试策略、状态监控全链路配套**

本文就想把这件事说清楚：从痛点、到原理、到工程化方案，再到可运行的示例代码。你会看到一个完整的闭环系统，告诉你如何让每天早上 8 点去抓“中国政府采购网”公告，既不重复，也不错漏。

为什么你的去重总是失效？真正的问题其实在“竞态”

本文探讨了分布式采集中的一致性问题，包括重复抓取和数据丢失，并提出了一个完整的解决方案。系统分为五个部分：定时触发、抓取列表、原子去重、任务流转和消费持久化。使用Redis SET和Lua脚本实现原子去重，Redis Stream作为可靠队列，以及数据库的幂等写入，确保数据一致性。文章还提供了示例代码，展示了如何实现这一系统。

后端

数据库

网络与通信

分布式采集系统如何解决一致性问题？本文详解从URL去重到任务流转的完整方案，包括Redis原子去重、Stream可靠队列、消费者组等工业级实现，提供可运行代码示例，确保数据不重复、不丢失。

消息队列

PostgreSQL

SQLite

MySQL

Redis

Kafka

Agent

SaaS

腾讯云数据库特惠，新用户首单0.5折起


2026新春采购季

tencentdb-catalog

postgres

message-queue-catalog

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云AI代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

功能1上新10个字符

功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符。

功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符。

功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符

功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符

功能4上新

文章&问答评论现已支持表情

全新交互，全新视觉，新增快捷键、悬浮工具栏、高亮块等功能并同时优化现有功能，全面提升创作效率和体验

社区富文本编辑器全新改版！诚邀体验～ 

精选全网热门MCP server，让你的AI更好用 🚀

💥开发者 MCP广场重磅上线！

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

为什么你的去重总是失效？真正的问题其实在“竞态”-腾讯云开发者社区-腾讯云

为什么你的去重总是失效？真正的问题其实在“竞态”

为什么你的去重总是失效？真正的问题其实在“竞态”

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐