如何用Python一键启动并监控多个requests爬虫

我知道 scrapy 可以使用 scrapyd 来监控可是我有很多 requests 写的爬虫该如何监控呢？我目前的做法是在 pycharm 里一个个去运行程序然后在 run 栏里会出来多个标签可以监控输出或者是打开多个 cmd 每个里面 python xxxx 但是这样当电脑重启再来启动多个爬虫很麻烦有没有好的方法或者 pycharm 现成的设置能让我配置好所有爬虫程序后一键启动并且分成多个标签让我来监控程序的输出呢？

songsunli 1楼

你可以直接写一个调用其他爬虫 main 函数的脚本，然后直接输出控制台就好了

wuwangju 2楼

要一键启动并监控多个requests爬虫，可以用concurrent.futures或multiprocessing。下面这个方案用ThreadPoolExecutor来并发执行，用队列来收集结果和异常。

import requests
import concurrent.futures
from queue import Queue
import time
from typing import List, Dict, Callable

class SpiderMonitor:
    def __init__(self, max_workers: int = 5):
        self.max_workers = max_workers
        self.result_queue = Queue()
        self.error_queue = Queue()
    
    def spider_task(self, url: str, spider_func: Callable) -> Dict:
        """单个爬虫任务"""
        try:
            result = spider_func(url)
            return {"url": url, "status": "success", "data": result}
        except Exception as e:
            return {"url": url, "status": "error", "error": str(e)}
    
    def run_spiders(self, urls: List[str], spider_func: Callable) -> Dict:
        """启动所有爬虫并监控"""
        results = {"success": [], "errors": []}
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # 提交所有任务
            future_to_url = {
                executor.submit(self.spider_task, url, spider_func): url 
                for url in urls
            }
            
            # 监控任务完成情况
            for future in concurrent.futures.as_completed(future_to_url):
                url = future_to_url[future]
                try:
                    result = future.result(timeout=10)
                    if result["status"] == "success":
                        results["success"].append(result)
                        print(f"✓ {url}: 爬取成功")
                    else:
                        results["errors"].append(result)
                        print(f"✗ {url}: {result['error']}")
                except concurrent.futures.TimeoutError:
                    error_result = {"url": url, "status": "error", "error": "请求超时"}
                    results["errors"].append(error_result)
                    print(f"✗ {url}: 请求超时")
        
        return results

# 示例爬虫函数
def sample_spider(url: str):
    """示例爬虫函数"""
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
    }
    response = requests.get(url, headers=headers, timeout=5)
    response.raise_for_status()
    return len(response.text)  # 返回页面长度作为示例

# 使用示例
if __name__ == "__main__":
    urls = [
        "https://httpbin.org/get",
        "https://httpbin.org/status/404",
        "https://httpbin.org/delay/2",
        "https://invalid-url-123.com",
        "https://httpbin.org/status/200"
    ]
    
    monitor = SpiderMonitor(max_workers=3)
    start_time = time.time()
    
    print("开始爬取...")
    results = monitor.run_spiders(urls, sample_spider)
    
    print(f"\n爬取完成，耗时: {time.time() - start_time:.2f}秒")
    print(f"成功: {len(results['success'])} 个，失败: {len(results['errors'])} 个")

这个方案的核心是ThreadPoolExecutor管理并发，as_completed监控完成状态。每个爬虫任务独立执行，错误不会影响其他任务。返回结果包含成功和失败的数据，方便后续处理。

如果要更详细的监控，可以加个进度条或者日志记录。用ProcessPoolExecutor能避免GIL限制，但进程间通信会复杂点。

简单说就是线程池加异常处理。