用 Dramatiq 构建分布式网页爬虫：官方示例源码解析与实战运行指南

后端任务调度【免费下载链接】dramatiqA fast and reliable background task processing library for Python 3.项目地址https://gitcode.com/gh_mirrors/dr/dramatiq点击查看免费下载Dramatiqdramatiq是一款面向 Python 3 的快速、可靠的分布式后台任务处理库。本篇文章围绕仓库中的官方示例 examples/crawler/README.md 与其实现源码 example.py完整讲解如何用 Dramatiq 实现一个基于任务自投递self-enqueueing的分布式网络爬虫读懂它的去重机制、抓取逻辑与容错配置并掌握从安装依赖、启动消息队列、运行 worker 到触发抓取的全流程实操。读完本文你将能独立复现这套示例并理解 Dramatiq 的 Actor、Broker、Worker、中间件等核心概念在真实场景中的协作方式。示例概览一个极简的分布式爬虫该示例的目标是用 Dramatiq 实现一个非常简单的分布式网页爬虫。它的核心设计并不复杂却完整地体现了 Dramatiq 的典型用法用dramatiq.actor把一个普通函数crawl(url)声明为异步任务Actor任务执行时通过 HTTP 抓取页面、用正则提取所有a href链接对每个新链接再次调用crawl.send(anchor)把下一跳任务投递回消息队列交给任意空闲的 worker 处理——这就是分布式广度优先遍历BFS的天然实现借助 Memcached 的原子add操作完成 URL 去重避免同一个页面被重复抓取通过max_retries3与time_limit10000两个 Actor 选项声明失败重试次数上限与单次任务执行时限。整个示例在 RabbitMQ 与 Redis 两种 broker 上均可运行通过环境变量REDIS1一键切换。运行前置条件与安装1. 安装消息队列服务Dramatiq 本身不负责消息传输需要依赖外部消息中间件。示例支持以下两者之一RabbitMQ官方推荐的消息 broker通过dramatiq[rabbitmq]安装依赖Redis备选方案通过dramatiq[redis]安装依赖。安装服务后需要在终端中启动它redis-server或rabbitmq-server。从实现看示例默认假设 broker 监听在本机默认端口Redis 为127.0.0.1:6379、库 0见 dramatiq/brokers/redis.py 对RedisBroker的说明RabbitMQ 默认连接127.0.0.1:5672见 dramatiq/broker.py 中get_broker()的默认参数。2. 安装 Dramatiq 与示例依赖按所选 broker 二选一安装 Dramatiqpip install dramatiq[rabbitmq] # 或者 pip install dramatiq[redis]再安装示例自身的依赖examples/crawler/requirements.txt 声明了两个包pip install -r requirements.txt其中pylibmc是 Python 的 Memcached 客户端示例用它做 URL 去重requests[security]是 HTTP 客户端带安全相关的额外依赖。3. 验证安装可在 Python 交互环境中确认 Dramatiq 已就绪import dramatiq print(dramatiq.__version__) # 当前仓库版本为 2.2.1见 dramatiq/__init__.py分步运行指南原文档给出了完整六步运行流程这里逐条展开并补充说明第 1 步安装 RabbitMQ 或 Redis。见上文二者选其一即可示例源码两种都兼容。第 2 步安装 Dramatiq。pip install dramatiq[rabbitmq] # 使用 RabbitMQ 时 pip install dramatiq[redis] # 使用 Redis 时第 3 步安装示例依赖。进入示例目录后执行pip install -r requirements.txt第 4 步启动消息队列服务。在终端窗口运行redis-server或rabbitmq-server并保持该进程运行。第 5 步启动 worker。在另一个独立的终端中运行dramatiq example这条命令会导入当前目录下的example模块即 examples/crawler/example.py并启动 Dramatiq worker 进程从队列中消费消息。若你希望使用 Redis broker需要在该命令前加上环境变量前缀REDIS1 dramatiq example第 6 步触发抓取。再开一个终端运行python example.py https://example.com程序会通过crawl.send(args.url)向队列投递第一条抓取任务随后 worker 开始抓取并不断派生出新任务。启动 worker 后终端会输出形如Dramatiq 2.2.1 is booting up.与Worker process is ready for action.的日志抓取过程中则能看到Crawling ......、Done crawling .... Found N anchors.以及重复 URL 的URL ... has already been visited. Skipping...提示。源码逐段解析模块级初始化去重客户端、链接正则与线程局部状态logger logging.getLogger(example) memcache_client pylibmc.Client([localhost], binaryTrue) memcache_pool pylibmc.ThreadMappedPool(memcache_client) anchor_re re.compile(rba href([^])) state local()memcache_poolpylibmc.ThreadMappedPool为每个线程维护独立的 Memcached 连接天然适配 Dramatiq 多线程 worker 模型anchor_re以字节串模式匹配a href...形式的链接后面直接对response.content字节流做finditerstate local()线程局部存储用于在每个线程内复用一个requests.Session见下文get_session()。通过环境变量切换 brokerif os.getenv(REDIS) 1: from dramatiq.brokers.redis import RedisBroker broker RedisBroker() dramatiq.set_broker(broker)设置REDIS1时使用 Redis broker否则使用 Dramatiq 的默认全局 broker。默认 broker 的选取逻辑见 dramatiq/broker.py优先尝试RabbitmqBroker若未安装 RabbitMQ 依赖则回退到RedisBroker()RedisBroker()无参数构造时默认连接本机 Redisredis://127.0.0.1:6379/0也支持RedisBroker(urlredis://...)或RedisBroker(host..., port..., db..., password...)等定制方式详见 dramatiq/brokers/redis.pydramatiq.set_broker(broker)将全局 broker 替换为指定实例后续所有dramatiq.actor声明与消息投递都走该 broker。线程局部 HTTP Session 复用def get_session(): session getattr(state, session, None) if session is None: session state.session requests.Session() return session每个 worker 线程首次调用时创建requests.Session并缓存在线程局部变量中之后复用。Session会复用 TCP 连接与 HTTP 连接池在批量抓取场景下能显著减少建连开销。声明爬虫 Actormax_retries 与 time_limit 的语义dramatiq.actor(max_retries3, time_limit10000) def crawl(url): ...这是整个示例中最能体现 Dramatiq 容错能力的两行配置max_retries3任务处理失败时最多自动重试 3 次之后消息进入失败dead-letter状态。Dramatiq 的 Retries 中间件 默认max_retries20、min_backoff15000毫秒15 秒、max_backoff86400000*7毫秒7 天会按指数退避重新入队Actor 级选项会覆盖中间件默认值见 dramatiq/middleware/retries.py 中message.options.get(max_retries, actor.options.get(max_retries, self.max_retries))的取值优先级time_limit10000单次任务最长执行 10000 毫秒10 秒超时会被 TimeLimit 中间件 在 worker 线程中触发TimeLimitExceeded异常中断。该中间件默认time_limit600000毫秒10 分钟、检查interval1000毫秒在 CPython 上通过向工作线程设置异步异常实现因此它无法中断阻塞的系统调用如陷入内核的 socket 读写这是其设计边界见 dramatiq/middleware/time_limit.py 的文档说明。这两个选项都属于 broker 的actor_options由中间件注册若在未加载对应中间件的 broker 上声明dramatiq.actor会抛出ValueError见 dramatiq/actor.py。默认 broker 已内置AgeLimit、TimeLimit、ShutdownNotifications、Callbacks、Pipelines、Retries六类中间件见 dramatiq/middleware/init.py。基于 Memcached 的 URL 去重url_hash hashlib.md5(url.encode(utf-8)).hexdigest() with memcache_pool.reserve() as client: added client.add(url_hash, b, time3600) if not added: logger.warning(URL %r has already been visited. Skipping..., url) return对 URL 取 MD5 作为键调用 Memcached 的add仅当键不存在时才写入成功等价于 SET NX 语义写入成功返回 True说明该 URL 首次遇到继续抓取写入失败说明已访问过直接跳过键的过期时间设为 3600 秒1 小时。这意味着去重不是永久的同一 URL 在 1 小时后可能被再次抓取。若要延长去重窗口可以调大time参数或改用 Redis 等带持久化的去重后端。抓取与链接提取logger.info(Crawling %r..., url) matches 0 session get_session() with closing(session.get(url, timeout(3.05, 5), streamTrue)) as response: if not response.headers.get(content-type, ).startswith(text/html): logger.warning(Skipping URL %r since its not HTML., url) return for match in anchor_re.finditer(response.content): anchor match.group(1).decode(utf-8) if anchor.startswith(http://) or anchor.startswith(https://): crawl.send(anchor) matches 1 logger.info(Done crawling %r. Found %d anchors., url, matches)要点timeout(3.05, 5)requests 的二元超时——前一个是连接超时 3.05 秒后一个是读取超时 5 秒防止单页抓取拖垮任务streamTrue流式获取响应配合content-type过滤只有text/html页面才会继续解析非 HTML 资源图片、CSS 等直接跳过避免浪费 workeranchor_re.finditer(response.content)直接在字节流上匹配a href...只投递绝对地址以http://或https://开头的链接相对路径链接/about、#anchor等被忽略——这是示例为保持简单而做的取舍crawl.send(anchor)是分布式爬虫的核心新链接被作为消息投回队列由任意 worker 进程/线程消费形成抓取 → 发现链接 → 再投递的接力即广度优先爬取。入口函数与命令行参数def main(args): parser argparse.ArgumentParser() parser.add_argument(url, typestr, helpa URL to crawl) args parser.parse_args() crawl.send(args.url) return 0 if __name__ __main__: sys.exit(main(sys.argv))main只做一件事把命令行传入的起始 URL 通过crawl.send(args.url)投递到队列。投递入口的底层调用链为Actor.send()→send_with_options()→broker.enqueue(message, delayNone)见 dramatiq/actor.pyRedis broker 的enqueue会给消息附加redis_message_id并通过 Lua 脚本写入队列见 dramatiq/brokers/redis.py。底层原理从投递到执行的完整消息流转结合源码一次抓取任务的完整生命周期如下投递crawl.send(url)构造Message包含队列名、Actor 名、参数等见 dramatiq/actor.py并交给 broker 的enqueue写入消息队列消费worker 中每个队列对应一个ConsumerThread调用 broker 的consume获取Consumer迭代器拉取消息Redis 实现见 dramatiq/brokers/redis.py 的_RedisConsumer支持预取prefetch与渐进式长轮询退避分发WorkerThread从共享优先队列取消息执行 Actor 函数见 dramatiq/worker.py中间件执行前后broker 按序触发before_process_message/after_process_message等钩子——Retries在失败时计算退避并重新入队TimeLimit登记/清除线程超时确认任务成功或重试耗尽后consumer 对消息执行ack成功移除或nack移入死信队列见 dramatiq/broker.py 的语义定义。Worker 运行模型dramatiq example背后发生了什么dramatiq example实际调用 dramatiq/cli.py 的入口由 dramatiq/main.py 转发其默认运行模型为进程数默认等于机器 CPU 核心数--processes/-p可调每进程线程数默认 8--threads/-t可调每个 worker 进程导入example模块、获取 broker、启动Worker(broker, worker_threads...)再通过Worker.start()启动消费者线程与工作线程见 dramatiq/cli.py。常用参数示例# 指定 2 个进程、每进程 4 线程 dramatiq --processes 2 --threads 4 example # 只监听特定队列 dramatiq example -Q default # 开发时监听文件变更自动重载需额外安装 dramatiq[watch] dramatiq --watch . example完整参数列表可通过dramatiq -h查看--worker-shutdown-timeout、--pid-file、--log-file等选项在 dramatiq/cli.py 中均有定义。常见问题与调优建议现象 / 需求说明与对策同一 URL 短时间内被重复抓取去重键 TTL 为 3600 秒可调大 example.py 中client.add(url_hash, b, time3600)的time生产环境可换用 Redis 去重并考虑持久化抓取慢导致任务超时被中断time_limit10000为 10 秒硬上限且 TimeLimit 无法中断阻塞的 socket 系统调用可调大该值并缩小 requests 的timeout页面含大量相对链接被忽略示例只投递绝对 URL可扩展链接归一化逻辑拼接urllib.parse.urljoin以抓取站内子页面任务反复失败max_retries3之后消息进入死信队列生产环境可配合Retry异常或throws选项声明不必重试的异常见 dramatiq/middleware/retries.py想限制抓取速率可接入 Dramatiq 的rate_limits模块如 dramatiq/rate_limits/bucket.py 的令牌桶限流器为crawl添加限流选项想横向扩容只需在更多机器上运行同样的dramatiq example消息由 broker 统一分发天然支持多机分布式抓取相关源码索引以下文件与本文内容直接相关便于深入阅读示例文档与实现examples/crawler/README.md、examples/crawler/example.py、examples/crawler/requirements.txtActor 声明与消息构造dramatiq/actor.pyBroker 抽象与全局实例dramatiq/broker.pyRedis / 测试用 broker 实现dramatiq/brokers/redis.py、dramatiq/brokers/stub.py重试与超时中间件dramatiq/middleware/retries.py、dramatiq/middleware/time_limit.pyCLI 与 Workerdramatiq/cli.py、dramatiq/worker.py其他官方示例examples/basic/example.py最简入门、examples/scheduling/example.py定时任务、examples/use_actors_from_different_files多文件 Actor 组织小结examples/crawler虽然只有约 70 行代码却浓缩了 Dramatiq 的核心用法Actor 声明与消息投递、broker 切换、中间件驱动的重试与超时控制、多进程多线程 worker 模型以及任务自投递这一实现分布式图遍历的经典模式。以此为起点你可以继续探索 docs/source/guide.rst 中的完整用户指南或参考 examples 目录下的调度、结果返回、组合任务等进阶示例把 Dramatiq 应用到更复杂的后台任务场景中。赞分享后端任务调度【免费下载链接】dramatiqA fast and reliable background task processing library for Python 3.项目地址https://gitcode.com/gh_mirrors/dr/dramatiq点击查看免费下载相关推荐dramatiq实战爬虫应用构建高性能分布式爬虫系统的终极指南dramatiq实战爬虫应用构建高性能分布式爬虫系统的终极指南 在当今数据驱动的时代高效获取网络信息至关重要。dramatiq作为Python生态中一款快速后端任务调度Crawlee BasicCrawler 实战指南基于 crawlee/basic 构建并行网页爬虫Crawlee BasicCrawler 实战指南基于 crawlee/basic 构建并行网页爬虫 导读 本文围绕 Crawlee 仓库中 package后端网页爬虫构建自己的网页爬虫gumbo-parser实战指南构建自己的网页爬虫gumbo parser实战指南 想要快速构建一个高效的网页爬虫系统吗gumbo parser作为一款纯C99语言编写的HTML5解析库后端上一篇Glance打造你的终极自托管信息聚合仪表板下一篇Deep-Live-Cam终极指南3分钟掌握实时换脸技术创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考