如何用调度器每8小时并发运行Scrapy项目中的100个爬虫？

针对你在同一个Scrapy项目里管理100个爬虫、需要每8小时并发运行且数据写入同一张表的需求，以下是分维度的最优实现方案：

一、基础调度：Cron + 自定义并发启动脚本 #

这是最轻量化、易维护的方案，适合绝大多数场景：

1. 编写爬虫批量启动脚本 #

用Python进程池控制并发数，避免一次性启动100个爬虫耗尽服务器资源（根据服务器配置，建议并发数设为10-20）：

import concurrent.futures
from scrapy.crawler import CrawlerProcess
from scrapy.utils.project import get_project_settings
from scrapy.spiderloader import SpiderLoader

def get_all_spider_names():
    """获取项目中所有爬虫名称"""
    settings = get_project_settings()
    loader = SpiderLoader(settings)
    return list(loader.list())

def run_single_spider(spider_name):
    """启动单个爬虫"""
    process = CrawlerProcess(get_project_settings())
    process.crawl(spider_name)
    process.start()

if __name__ == "__main__":
    spider_list = get_all_spider_names()
    # 控制并发进程数，根据服务器CPU/内存调整
    with concurrent.futures.ProcessPoolExecutor(max_workers=15) as executor:
        executor.map(run_single_spider, spider_list)

2. 配置Cron定时任务 #

通过系统Cron实现每8小时触发一次脚本：

# 编辑Cron任务
crontab -e

# 添加以下内容（替换项目路径和日志路径）
0 */8 * * * cd /your/scrapy/project/path && python run_spiders.py >> crawl_combined.log 2>&1

日志会记录所有爬虫的运行状态，方便后续排查问题。

二、数据库写入优化（核心） #

因为所有爬虫共享一张表，必须解决并发写入的性能和冲突问题：

1. 批量插入Pipeline #

在Scrapy的Pipeline中实现批量插入，减少数据库连接开销：

import pymysql
from DBUtils.PooledDB import PooledDB
from scrapy.exceptions import DropItem

class BatchDBPipeline:
    def __init__(self):
        # 初始化数据库连接池，避免频繁创建销毁连接
        self.db_pool = PooledDB(
            creator=pymysql,
            maxconnections=20,
            mincached=5,
            host="your_db_host",
            user="your_db_user",
            password="your_db_pwd",
            database="your_db_name",
            charset="utf8mb4"
        )
        self.batch_size = 100  # 每积累100条数据插入一次
        self.item_buffer = []

    def process_item(self, item, spider):
        self.item_buffer.append(dict(item))
        if len(self.item_buffer) >= self.batch_size:
            self._insert_batch()
            self.item_buffer = []
        return item

    def close_spider(self, spider):
        # 爬虫结束时插入剩余数据
        if self.item_buffer:
            self._insert_batch()

    def _insert_batch(self):
        conn = self.db_pool.connection()
        try:
            cursor = conn.cursor()
            # 生成插入SQL，自动适配Item字段
            fields = ", ".join(self.item_buffer[0].keys())
            placeholders = ", ".join(["%s"] * len(self.item_buffer[0]))
            # 加入去重更新逻辑（根据你的唯一键调整）
            sql = f"""
                INSERT INTO target_table ({fields}) VALUES ({placeholders})
                ON DUPLICATE KEY UPDATE {", ".join([f"{k}=VALUES({k})" for k in self.item_buffer[0].keys()])}
            """
            # 批量执行插入
            values = [tuple(item.values()) for item in self.item_buffer]
            cursor.executemany(sql, values)
            conn.commit()
        except Exception as e:
            conn.rollback()
            raise DropItem(f"批量插入失败: {str(e)}")
        finally:
            cursor.close()
            conn.close()

2. 数据库层面优化 #

• 给表添加唯一索引（比如爬虫ID+数据主键），避免重复数据
• 开启数据库的批量写入优化（比如MySQL的innodb_flush_log_at_trx_commit=2，适合非强实时场景）
• 若使用PostgreSQL，可改用copy_from方式，比executemany性能提升数倍

三、进阶调度：Celery + 定时任务（适合复杂场景） #

如果需要更灵活的任务监控、重试机制，推荐用Celery：

1. 将每个爬虫的启动逻辑包装成Celery任务
2. 用Celery Beat设置每8小时触发所有爬虫任务
3. 通过Celery Worker控制并发数，用Flower监控任务状态

这种方案适合需要频繁调整调度规则、或需对失败爬虫自动重试的场景。

四、关键注意事项 #

• 资源限制：绝对不要一次性启动100个爬虫，根据服务器CPU核心数（比如8核服务器跑10-15个并发）调整，避免OOM或带宽耗尽
• 日志隔离：可修改Scrapy日志配置，让每个爬虫生成独立日志文件，方便定位单个爬虫的问题
• 失败重试：在启动脚本中加入重试逻辑，比如对启动失败的爬虫重试2-3次
• 数据库锁：批量插入时尽量缩短事务时间，避免长时间占用表锁影响其他爬虫写入

内容的提问来源于stack exchange，提问作者fractal397