企业官网建设流程全解析

做爬虫项目，任务一多，你就会开始想一个事：列表页抓完了才能抓详情页，详情页入库之后才能做数据清洗，这些环节怎么串起来自动跑？我最早是写一个main.py，从上到下硬编码顺序执行，跑一会儿崩了，从断点接着跑又得手动改代码。后来也试过部署一套完整的调度框架，结果光维护调度器本身的时间就快赶上写业务代码了。最后我花了大半天，用大概100行Python写了一个本地化的Mini-Airflow，任务依赖、失败重试、执行日志全都有了。这篇文章就把这套方案完整拆开讲，代码可以直接抄走改，适合中小型爬虫项目、脚本自动化、数据采集管线的同学参考。

1. 为什么自己写调度器，而不是直接上Airflow

1.1 调度器到底解决了爬虫里的什么问题

爬虫项目看着简单，实际落地时最耗时间的不是单个爬虫函数怎么写，而是多个任务怎么编排。一个典型的采集流程通常是：抓取列表页拿URL，再去请求详情页解析字段，最后把数据清洗后写入数据库或文件。这三步天然有先后顺序，详情页的请求依赖列表页的结果，清洗入库又依赖详情页的结果。如果你只有一个脚本，手动维护顺序，第一次跑没问题，第二次多了增量任务、失败重试，代码就开始变成一坨if else。

调度器解决的正是这种"任务编排"问题。它把每个业务步骤抽象成一个独立任务，任务之间声明依赖关系，由调度器统一决定执行顺序。A没跑完，B就不该启动；B失败了，要么重试要么标记失败，而不是让下游无脑继续跑。这个思路在数据工程里叫DAG调度，Airflow是其中的行业标准，但爬虫项目往往不需要那么重的机制。

1.2 工业级调度器的"重"和Mini方案从哪里切入

Airflow专业、强大，有Web界面、定时调度、分布式执行器、丰富的监控告警，但对一个跑在本地的爬虫项目来说，部署成本是真不低。你要装Python依赖、初始化数据库、配置Executor、启动Scheduler和Web Server，一套下来至少多花半天时间。而且Airflow的任务实例状态存数据库，和爬虫代码放一起反而让项目变臃肿。

我需要的其实只有三个能力：任务依赖、失败重试、日志记录。这三个能力本质上不需要任何外部依赖，用Python原生的dict、set、logging就能实现。100行代码比引入一套框架更容易维护，出了问题自己就能排查。这也是Mini-Airflow这类"小工具"存在的价值：复杂工具解决复杂问题，但普通人遇到的大部分问题其实用不上那套复杂度。

1.3 什么时候该用这类Mini方案，什么时候别硬凑

Mini方案适合任务数量在几个到十几个之间、任务之间有明确的先后关系或并行关系不大、跑在单机上的场景。比如爬虫列表页→详情页→入库，或者每日定时统计数据、生成报表，再或者几个爬虫脚本之间互相补充数据。这种场景下，你需要的只是在一个进程里把所有任务跑完，跑完就走。

如果你的任务数量到了几十个、需要跨机器调度、需要挂起和恢复历史任务状态、需要多人协作维护复杂的调度拓扑，那还是老老实实用Airflow或Prefect吧。Mini方案没有持久化状态，进程重启后一切从头来，它定位是"轻量、够用、贴近业务"，不是替代专业调度器。

2. 设计思路：任务依赖的本质是DAG

2.1 核心抽象：任务、依赖图、执行器

实现一个小型调度器，第一步是抽象出三个东西：任务（Task）、依赖图（Graph）、执行器（Executor）。

任务是业务函数本身，带上任务ID、重试次数、重试间隔这些元信息。依赖图记录每个任务依赖谁，底层用邻接表表示：一个dict，key是任务ID，value是这个任务依赖的任务ID集合。执行器负责按照依赖图算出执行顺序，逐个运行任务，运行失败就按策略重试。

这和你爬山时的路线规划很像。多个景点之间有先后关系，你得先到山口才能去山腰，到了山腰才能去山顶。那地图本身就是一个图结构，你的行走顺序就是对这个图做拓扑排序。

2.2 拓扑排序：怎么知道A得在B前面跑

有向无环图（DAG）里，拓扑排序能保证每个节点都在它的所有前置节点之后出现。对于依赖关系"B依赖A"，排序结果里A一定在B前面。这样执行器只需要按排序后的列表顺序跑，任务依赖天然满足。

拓扑排序有两种经典算法：Kahn算法和DFS。对于这个场景，Kahn算法更好理解，思路是：先统计每个任务的入度，入度为0的任务没有前置依赖，可以立即执行；执行完一个任务后，把依赖它的所有任务的入度减1，如果减到0就放进可执行队列；循环直到所有任务都排进去。如果最后还有任务没排进去，说明图里有环，这就是"检测到循环依赖"的逻辑基础。

2.3 功能取舍：为什么"少做"反而好用

写这个Mini调度器时，我反复提醒自己一件事：不要顺手把功能越加越多。网上有很多开源的迷你调度器，写着写着就加了并发执行、定时触发、任务状态持久化、甚至Web UI，代码量翻倍，调试复杂度也跟着翻倍。一旦加了并发，你就得处理线程安全、共享变量、任务间竞态，这些坑比业务代码还难搞。

我最终做的取舍是：纯单线程顺序执行，不加并发；不做持久化，进程结束状态就丢；不做定时触发，只做手动触发或配合系统级定时任务（比如crontab）；不做Web界面，日志就是最好的观测手段。这四个"不做"换来的是代码短、可读性强、出问题只需要看几十行就能定位。这种取舍在真实项目里非常重要，功能边界清晰，比功能多但没人敢改的代码强得多。

2.4 任务之间怎么共享数据

任务之间往往不是完全独立的。详情页解析需要列表页抓到的URL列表，入库需要详情页解析出的字段。那任务之间怎么传递数据？

最简单的方案是让任务函数返回值，由调度器把返回值存到一个共享的上下文字典里。下游任务执行时，可以通过任务ID拿到上游任务的返回值。这就够了。不需要引入复杂的参数注入机制，一个context字典打天下。具体实现是在执行器里维护一个results字典，任务跑完后把返回值存进去，执行下一个任务前先检查依赖任务是否都在results里，如果不在就直接报错，防止"依赖没跑完就往下走"的情况。

3. 完整代码实现与逐段拆解

3.1 基础设施：日志与任务注册

直接上代码，先看整体结构。我把核心类命名为MiniAirflow，全部代码放在一个文件里。

import time import traceback import logging from collections import defaultdict, deque from typing import Callable, Dict, Any, Set class MiniAirflow: def __init__(self, name="mini_airflow", log_file=None): self.name = name self.tasks: Dict[str, dict] = {} self.dependencies: Dict[str, Set[str]] = defaultdict(set) self.results: Dict[str, Any] = {} self.logger = self._setup_logger(log_file)

这段做了三件事：初始化任务字典、初始化依赖表、初始化日志器。tasks字典里存的是任务ID到任务配置的映射，配置包括业务函数、重试次数、重试间隔。dependencies用defaultdict(set)来存依赖关系，好处是直接add不需要先判断key是否存在。results是任务结果共享的context，后面数据传递靠它。

看日志器初始化，我不建议直接用全局的logging，而是给每个调度器实例建独立的logger，这样日志可以写入不同文件，互不干扰。

def _setup_logger(self, log_file): logger = logging.getLogger(self.name) logger.setLevel(logging.INFO) fmt = logging.Formatter( "%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S" ) if log_file: fh = logging.FileHandler(log_file, encoding="utf-8") fh.setFormatter(fmt) logger.addHandler(fh) sh = logging.StreamHandler() sh.setFormatter(fmt) logger.addHandler(sh) return logger

这里有个细节：如果重复运行脚本，FileHandler会不断往同一个文件追加日志，容易混乱。我实际使用时会按时间戳生成日志文件名，比如crawler_20250101_100000.log，每次运行一个文件，排查问题很方便。

3.2 任务装饰器与依赖注册

为了让代码用起来像Airflow那样优雅，我用装饰器注册任务。每个业务函数只要加上@mini.task(...)，就自动被登记到调度器里。

def task(self, task_id: str, retries: int = 3, retry_delay: int = 5): def decorator(func: Callable): self.tasks[task_id] = { "func": func, "retries": retries, "retry_delay": retry_delay, } return func return decorator def set_dependency(self, task_id: str, depends_on: str): self.dependencies[task_id].add(depends_on)

task这个装饰器接收三个参数：task_id是任务唯一标识，retries是最大重试次数，retry_delay是失败后等待的秒数。这样业务代码里只要写清楚任务ID和重试策略就行，执行细节全部交给调度器。

set_dependency是声明依赖的核心方法，含义是"task_id这个任务要跑，必须先跑完depends_on"。这里我用depends_on这个参数名，比parent或者upstream更直白。实际使用时，一个任务可以依赖多个上游任务，比如C依赖A和B，那就要调用两次set_dependency。

3.3 拓扑排序与循环依赖检测

排序部分我用Kahn算法的简化版本。先统计每个任务的入度，再按入度从0开始往出排。

def _topological_sort(self) -> list: in_degree = {tid: 0 for tid in self.tasks} for tid, deps in self.dependencies.items(): for dep in deps: if dep not in self.tasks: raise KeyError(f"依赖的任务 {dep} 不存在") in_degree[tid] += 1 queue = deque([tid for tid in self.tasks if in_degree[tid] == 0]) order = [] while queue: node = queue.popleft() order.append(node) for tid, deps in self.dependencies.items(): if node in deps: in_degree[tid] -= 1 if in_degree[tid] == 0: queue.append(tid) if len(order) != len(self.tasks): raise RuntimeError("检测到循环依赖，任务无法正常排序") return order

注意两个地方。第一，依赖的任务必须存在，如果写了个set_dependency("fetch_detail", "fetch_list")，但fetch_list这个任务ID并不存在，这里会抛出KeyError，帮你尽早发现配置错误。第二，如果存在循环依赖，比如A依赖B、B又依赖A，那么两个节点的入度永远不可能变成0，最后order的长度会小于tasks的总数。这一步的判断是整个调度器的安全性保障。

为了让大家直观理解，我加了一个简化版的依赖校验和拓扑排序测试。假设任务A、B、C，C依赖A和B，那in_degree初始为{A:0, B:0, C:2}，队列先弹A或B，执行完A后C的入度减为1，再执行B后C入度减为0，被加入队列，最终顺序是["A", "B", "C"]，任意方向都满足依赖关系。

3.4 执行引擎：失败重试与结果收集

核心的run方法就是把拓扑排序结果和任务执行串起来，做失败重试、结果收集、日志记录。完整代码如下：

def run(self): self.logger.info(f"调度器 {self.name} 开始运行") order = self._topological_sort() self.logger.info(f"任务执行顺序: {' -> '.join(order)}") for tid in order: deps = self.dependencies.get(tid, set()) missing = [d for d in deps if d not in self.results] if missing: self.logger.error(f"任务 {tid} 的依赖 {missing} 未成功执行，跳过") raise RuntimeError(f"依赖未满足: {tid} 缺少 {missing}") task_info = self.tasks[tid] func = task_info["func"] retries = task_info["retries"] delay = task_info["retry_delay"] for attempt in range(1, retries + 1): try: start = time.time() self.logger.info(f"开始执行任务 [{tid}] (第 {attempt}/{retries} 次尝试)") result = func(self.results) elapsed = time.time() - start self.results[tid] = result self.logger.info(f"任务 [{tid}] 执行成功, 耗时 {elapsed:.2f} 秒") break except Exception as e: self.logger.error(f"任务 [{tid}] 执行失败: {e}") if attempt < retries: self.logger.info(f"任务 [{tid}] 将于 {delay} 秒后重试") time.sleep(delay) else: self.logger.error( f"任务 [{tid}] 重试 {retries} 次后仍失败\n{traceback.format_exc()}" ) raise self.logger.info("全部任务执行完毕") return self.results

这里有两个设计细节值得展开。第一，把self.results作为参数传给了任务函数，下游任务就能通过self.results["上游任务ID"]拿到上游结果。这个设计比Airflow的XCom简单，但足够通用。第二，重试策略是"串行重试"，即失败后sleep几秒再跑，对爬虫场景特别合适，因为很多失败是网络抖动或目标站点临时限流，隔几秒重试就能恢复。如果任务本身有内部状态，也可以在上游为每个任务设置不同的重试参数。

整个调度器的核心逻辑到这里就结束了，从导入到头尾完全统计，代码量大致在100行左右。对比Airflow动辄上万的代码量，这个体量对爬虫项目非常友好。

4. 实操过程：跑通一套真实的爬虫调度

4.1 设计任务链

我先用一个电商商品采集的例子来演示。任务链设计为四步：

• fetch_category：获取分类页，模拟请求分类列表，返回分类链接列表。
• fetch_product_list：根据分类链接抓商品列表页，返回商品详情页URL列表。
• fetch_product_detail：逐个请求详情页，解析价格、标题、描述。
• save_to_database：把解析结果整理后写入文件或数据库。

这四个任务有明确的依赖关系：fetch_product_list依赖fetch_category，fetch_product_detail依赖fetch_product_list，save_to_database依赖fetch_product_detail。同时，fetch_category没有上游，可以最先执行；save_to_database没有下游，是最后一个执行。

用set_dependency配置依赖时，我会给每个任务配置不同的重试策略。网络请求类任务（列表页、详情页）重试次数多一些，重试间隔也长一些，因为网络抖动是偶发的；数据库写入类任务重试次数少一些，因为如果是数据本身有问题，重试再多也没用。

4.2 组装完整代码

下面是把前面的调度器和具体爬虫任务拼接在一起的完整示例。实际项目里，你只需要替换任务函数内的逻辑，调度器部分完全不用改。

import random import time from mini_airflow import MiniAirflow mini = MiniAirflow("shop_crawler", log_file="shop_crawler.log") @mini.task("fetch_category", retries=3, retry_delay=2) def fetch_category(ctx): """ 模拟抓取分类页，返回分类链接列表 正常情况应该是 requests.get + BeautifulSoup 解析 """ print("抓取分类页...") time.sleep(0.5) return [f"https://shop.example.com/category/{i}" for i in range(1, 4)] @mini.task("fetch_product_list", retries=3, retry_delay=5) def fetch_product_list(ctx): categories = ctx["fetch_category"] print(f"根据 {len(categories)} 个分类抓取商品列表...") product_urls = [] for cat in categories: time.sleep(0.3) product_urls.extend([ f"{cat}/product/{j}" for j in range(1, 4) ]) return product_urls @mini.task("fetch_product_detail", retries=2, retry_delay=10) def fetch_product_detail(ctx): product_urls = ctx["fetch_product_list"] print(f"抓取 {len(product_urls)} 个商品详情...") results = [] for url in product_urls: time.sleep(0.2) # 模拟随机失败一次，验证重试机制 if random.random() < 0.15: raise RuntimeError(f"请求失败: {url}") results.append({ "url": url, "title": f"商品-{url[-1]}", "price": random.randint(10, 999) }) return results @mini.task("save_to_database", retries=1, retry_delay=3) def save_to_database(ctx): products = ctx["fetch_product_detail"] print(f"保存 {len(products)} 条商品数据...") with open("products.txt", "w", encoding="utf-8") as f: for p in products: f.write(f"{p['title']} | {p['price']} | {p['url']}\n") return f"saved {len(products)} items" mini.set_dependency("fetch_product_list", "fetch_category") mini.set_dependency("fetch_product_detail", "fetch_product_list") mini.set_dependency("save_to_database", "fetch_product_detail") mini.run()

任务函数统一接收ctx这个参数，也就是调度器内部维护的results字典。每个任务通过ctx["上游任务ID"]拿上游数据。这样代码结构清晰，任务之间没有任何隐式共享变量，符合可测试、可维护的要求。

这里random.random() < 0.15模拟了详情页抓取时的随机失败，用来验证重试机制的效果。配合日志，你能清楚看到哪个任务在哪个时间点失败了多少次、第几次成功。

4.3 运行结果与日志解读

运行上面的代码，控制台会输出类似下面的日志：

2025-01-01 10:00:01 [INFO] 调度器 shop_crawler 开始运行 2025-01-01 10:00:01 [INFO] 任务执行顺序: fetch_category -> fetch_product_list -> fetch_product_detail -> save_to_database 2025-01-01 10:00:01 [INFO] 开始执行任务 [fetch_category] (第 1/3 次尝试) 2025-01-01 10:00:01 [INFO] 任务 [fetch_category] 执行成功, 耗时 0.51 秒 2025-01-01 10:00:02 [INFO] 开始执行任务 [fetch_product_list] (第 1/3 次尝试) ... 2025-01-01 10:00:05 [ERROR] 任务 [fetch_product_detail] 执行失败: 请求失败: https://shop.example.com/category/2/product/2 2025-01-01 10:00:06 [INFO] 任务 [fetch_product_detail] 将于 10 秒后重试 2025-01-01 10:00:16 [INFO] 开始执行任务 [fetch_product_detail] (第 2/2 次尝试) 2025-01-01 10:00:19 [INFO] 任务 [fetch_product_detail] 执行成功, 耗时 2.95 秒 ... 2025-01-01 10:00:21 [INFO] 全部任务执行完毕

日志的关键价值在于可复盘。可以看出每一次重试的开始时间、耗时、失败原因，甚至可以按任务ID过滤日志来追踪单个任务的多次尝试。如果你用log_file参数指定了日志文件，那么控制台和文件都会有同步输出。

这里我建议把日志级别设为INFO，任务里的print只能到控制台，不会进日志文件。所以业务函数里如果有重要的输出，可以用logging.getLogger("shop_crawler").info(...)，或者统一把print改成ctx.logger之类的自定义输出，保证日志文件内容完整。

4.4 扩展：任务结果持久化与断点恢复

Mini方案虽然不持久化任务状态，但有一个非常实用的扩展：把self.results定期用json或pickle写进本地文件。如果任务链很长，跑了一半进程崩溃了，下次启动时可以先加载上次的结果，跳过已经成功的任务。

这个扩展不需要改调度器核心，只要在run方法里加几行：

import json, os def _save_checkpoint(self, tid): tmp = {k: v for k, v in self.results.items() if isinstance(v, (str, int, list, dict))} with open(f"{self.name}_checkpoint.json", "w", encoding="utf-8") as f: json.dump(tmp, f, ensure_ascii=False, indent=2) def _load_checkpoint(self): path = f"{self.name}_checkpoint.json" if os.path.exists(path): with open(path, "r", encoding="utf-8") as f: self.results.update(json.load(f))

我实际使用中，遇到网络爬虫中途断开，断点恢复能省掉重爬大量页面。不过要提醒一下：results里的数据如果是自定义对象，json序列化不了，这种情况需要用pickle，或者只持久化任务状态、不持久化数据本身。

5. 常见问题与排查技巧实录

5.1 循环依赖为什么一言不合就出现

循环依赖是写调度配置时最容易犯的错，尤其是任务多了以后。比如你写的时候想着"清洗依赖入库，入库也依赖清洗"，两个任务互相等待，拓扑排序就会报错。我的排查经验是：先打印出所有set_dependency调用，沿着依赖链画一遍，只要发现从某个任务出发能绕一圈回来，那就是环。

实际操作中，我会在set_dependency里加一条校验：不允许任务依赖自己，这是最常见的隐性环。如果配置了A -> B -> C -> A这种链条，我的_topological_sort会抛出RuntimeError，并把参与任务总数和已排序任务数都打出来，方便定位。

5.2 重试参数到底怎么设置才合理

很多教程告诉你"失败重试3次"，但实际场景里重试策略要分任务来定。网络请求类的任务，重试间隔长一点（5到30秒），给目标站点留出恢复时间；数据库写入类的任务，间隔短一点（2到5秒），重试次数少一点，因为大概率是数据问题。我自己常用的策略是：请求类retries=3, retry_delay=10，解析类retries=2, retry_delay=3，入库类retries=1, retry_delay=2。

重试不是越多越好。如果目标站点已经封了你的IP，重试10次也是白搭。所以我的调度器没有设计指数退避，而是固定间隔重试，因为爬虫场景下简单的策略往往最可控。如果你想做指数退避，在run方法里把sleep(delay)改成sleep(delay * attempt)就行，加了指数退避后重试效果会更有节奏感。

5.3 PyCharm里只显示Process finished with exit code 0，没输出内容怎么办

这个现象太常见了。你点运行，控制台只显示一行Process finished with exit code 0，中间啥内容都没有。出现这个，大部分情况是程序确实跑完了，但输出被吞了或者根本没执行到你预期的地方。

在调度器这个场景里，常见原因有三个：一是日志级别设置太高，比如logger.setLevel(logging.ERROR)，INFO级别的日志自然不显示；二是任务函数内部用了print，但调度器代码里加了日志handler覆盖了输出流，导致print的内容没刷出来；三是你有多个logger实例，日志写到了文件而不是控制台。

我的排查顺序是：先看log_file指定的日志文件里有没有内容，有内容就说明代码跑了，只是控制台没显示；然后在任务函数第一行加一个print("task start")，看是否执行到；最后检查logger配置，确认StreamHandler确实被添加了。如果还不行，把logger.setLevel(logging.DEBUG)，看看是不是有早期异常被吞了。

5.4 日志重复输出和中文乱码问题

日志重复输出，主要原因是你重复执行脚本时，前一次运行的FileHandler没有释放。用PyCharm反复运行，同一个logger名字会挂多个handler，导致日志打两遍。解决办法是在_setup_logger里先清空已有handler：

logger.handlers.clear()

或者用logger.removeHandler(handler)精确清理。这个细节别看小，实际调试时能省不少时间。

中文乱码问题几乎都是文件编码没指定。我在初始化logging.FileHandler时明确写了encoding="utf-8"，但Windows上默认编码是gbk，如果没指定encoding，中文日志就会变成乱码。这个问题排查起来很隐蔽，因为错误信息里根本不会提示编码问题。如果你在Windows上遇到乱码，优先检查FileHandler的编码参数，其次检查日志文件是用什么编码打开的。

另外提醒一下，在爬虫任务里如果要把数据写入文件，也一定要带encoding="utf-8"，否则Windows下写入中文会报UnicodeEncodeError。这也是新手写爬虫最容易遇到的问题。

5.5 任务函数改了，日志却显示旧代码在跑

这个坑我踩过好几次。Python的模块缓存机制会让旧的.pyc文件被重新使用，尤其是PyCharm里热更新不彻底的时候。遇到修改代码后日志输出还是旧逻辑，先重启解释器，再清一遍__pycache__目录。在命令行跑一下find . -name "*.pyc" -delete，问题基本就解决了。如果你已经改成了"重启大法"还是不行，那就要检查是不是有多个Python环境，跑的时候用的不是当前项目解释器。

6. 后续扩展：让Mini调度器更贴近实际项目

我这个Mini方案在项目里跑了一段时间之后，稳定性和维护成本都非常理想。在此基础上，你还可根据实际需求做几个小扩展。

• 加定时触发：在main.py里用while True + time.sleep循环判断当前时间是否到达目标时间点，到了就调用mini.run()。或者直接配系统crontab，更省事。
• 加并发执行：把所有独立的"入度为0"的任务丢到线程池里并行执行。要处理共享变量和异常收集，代码量会增加，但如果你有几组互不依赖的爬虫任务，提速效果很明显。
• 加Webhook通知：任务失败时用requests.post把失败信息发到企业微信、钉钉或邮件。接入很轻量，对长时间运行的脚本尤其有用。

我个人在实际项目里，目前这套Mini-Airflow承担了三个爬虫项目的数据采集任务。相比之前每个项目各自硬编码执行顺序，统一调度之后，新增任务只要加一个函数+注册依赖，几乎不用动其他代码。任务失败了看日志就能定位，日志文件按时分秒命名，归档和排查都很方便。如果你也需要编排多个爬虫任务，不妨从这套100行的方案开始，它简单到让你可以把精力集中在业务逻辑本身，而不是调度器上的各种新坑。