第4章 Python数据采集框架

4、Python数据采集框架：多线程/多进程、队列管理与数据落盘

好，咱们进入实战环节。数据采集这件事，说白了就是跟时间赛跑。你想想看，单线程爬一个页面要等几百毫秒，要是采集几百万条数据，那得等到猴年马月去？我刚开始做数据工程时，就吃过这个亏——用单线程跑了三天三夜，结果中间网络抖了一下，全白干了。

所以这一章，咱们重点聊聊怎么用Python把采集速度提上去。核心就四个字：并行采集。

4.1 多线程 vs 多进程：到底选哪个？

很多新手会纠结这个问题。我直接说结论：IO密集型任务用多线程，CPU密集型任务用多进程。

数据采集属于典型的IO密集型——大部分时间都在等网络响应、等磁盘写入。这时候多线程就够用了。Python的GIL（全局解释器锁）在IO等待时会自动释放，所以多线程能跑满带宽。

但如果你要做数据清洗、特征提取这类计算密集的操作，那就得上多进程了。我在项目中遇到过这种情况：采集回来的原始数据需要做实时解析，CPU占用直接飙到100%，多线程反而比单线程还慢。

  核心原则：
  网络请求、文件读写 → 多线程
数据解析、计算转换 → 多进程
混合场景 → 线程+进程组合使用

4.2 实战：多线程采集框架

先看一个最基础的多线程采集模板。我个人习惯用concurrent.futures模块，比手动管理threading要省心得多。

import concurrent.futures
import requests
import time

def fetch_url(url):
    """单个采集任务"""
    try:
        resp = requests.get(url, timeout=10)
        return resp.text
    except Exception as e:
        print(f"采集失败: {url} - {e}")
        return None

def multi_thread_collect(urls, max_workers=10):
    """多线程采集入口"""
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有任务
        future_to_url = {executor.submit(fetch_url, url): url for url in urls}
        
        # 收集结果
        for future in concurrent.futures.as_completed(future_to_url):
            url = future_to_url[future]
            try:
                data = future.result()
                if data:
                    results.append((url, data))
                    print(f"完成: {url}")
            except Exception as e:
                print(f"异常: {url} - {e}")
    return results

嗯，这里要注意：max_workers不是越大越好。我曾经把线程数设到100，结果目标网站直接把我IP封了。一般建议控制在5-20之间，具体要看目标网站的承受能力。

4.3 队列管理：生产者-消费者模式

单靠多线程还不够。你想想看，如果采集速度比处理速度快，内存就会爆掉。这时候就需要队列来缓冲。

Python的queue.Queue是线程安全的，我特别喜欢用它来做任务调度。看这个经典的生产者-消费者模型：

import queue
import threading
import time

# 全局任务队列
task_queue = queue.Queue(maxsize=1000)

def producer(urls):
    """生产者：不断往队列里放任务"""
    for url in urls:
        task_queue.put(url)
        print(f"放入队列: {url}")
    # 放一个结束信号
    task_queue.put(None)

def consumer():
    """消费者：从队列取任务并采集"""
    while True:
        url = task_queue.get()
        if url is None:  # 收到结束信号
            task_queue.task_done()
            break
        # 执行采集
        data = fetch_url(url)
        if data:
            # 这里可以写入缓冲区或直接落盘
            print(f"采集完成: {url}")
        task_queue.task_done()

# 启动
producer_thread = threading.Thread(target=producer, args=(url_list,))
consumer_threads = [threading.Thread(target=consumer) for _ in range(5)]

producer_thread.start()
for t in consumer_threads:
    t.start()

producer_thread.join()
task_queue.join()  # 等待所有任务完成

我的经验：队列大小一定要设置上限。如果不设maxsize，生产者跑得比消费者快时，内存会无限增长。我曾经在采集电商数据时没设上限，结果跑了半小时内存就爆了，数据全丢了。

4.4 数据缓冲：别让磁盘成为瓶颈

数据落盘也是个讲究活。每条数据都直接写文件，磁盘IO会成为新的瓶颈。我的做法是：先缓冲，再批量写入。

看这个缓冲区的实现：

import threading
import json
import time

class DataBuffer:
    """数据缓冲区，批量写入磁盘"""
    def __init__(self, file_path, buffer_size=100, flush_interval=5):
        self.file_path = file_path
        self.buffer_size = buffer_size
        self.flush_interval = flush_interval
        self.buffer = []
        self.lock = threading.Lock()
        self.last_flush_time = time.time()
        
        # 启动定时刷新线程
        self._start_flush_timer()
    
    def add(self, data):
        """添加数据到缓冲区"""
        with self.lock:
            self.buffer.append(data)
            if len(self.buffer) >= self.buffer_size:
                self._flush()
    
    def _flush(self):
        """实际写入磁盘"""
        if not self.buffer:
            return
        with open(self.file_path, 'a', encoding='utf-8') as f:
            for item in self.buffer:
                f.write(json.dumps(item, ensure_ascii=False) + '\n')
        self.buffer.clear()
        self.last_flush_time = time.time()
    
    def _start_flush_timer(self):
        """定时强制刷新"""
        def timer_loop():
            while True:
                time.sleep(self.flush_interval)
                with self.lock:
                    if time.time() - self.last_flush_time >= self.flush_interval:
                        self._flush()
        t = threading.Thread(target=timer_loop, daemon=True)
        t.start()
    
    def close(self):
        """关闭时强制刷新所有数据"""
        with self.lock:
            self._flush()

避坑指南：我曾经在缓冲区里只设了数量阈值，没设时间阈值。结果采集任务突然中断，缓冲区里还有几百条数据没来得及写入，全丢了。所以一定要加定时刷新机制。

4.5 完整框架：把一切串起来

好了，现在我们把多线程、队列、缓冲区整合成一个完整的采集框架：

class CollectPipeline:
    """完整的数据采集流水线"""
    def __init__(self, urls, output_file, 
                 num_workers=10, buffer_size=100):
        self.urls = urls
        self.output_file = output_file
        self.num_workers = num_workers
        self.task_queue = queue.Queue(maxsize=500)
        self.buffer = DataBuffer(output_file, buffer_size)
        self.stats = {'total': 0, 'success': 0, 'fail': 0}
        self.stats_lock = threading.Lock()
    
    def _worker(self):
        """工作线程"""
        while True:
            url = self.task_queue.get()
            if url is None:
                self.task_queue.task_done()
                break
            try:
                data = fetch_url(url)
                if data:
                    self.buffer.add({'url': url, 'data': data})
                    with self.stats_lock:
                        self.stats['success'] += 1
                else:
                    with self.stats_lock:
                        self.stats['fail'] += 1
            except Exception as e:
                print(f"工作线程异常: {e}")
            finally:
                self.task_queue.task_done()
                with self.stats_lock:
                    self.stats['total'] += 1
    
    def run(self):
        """启动流水线"""
        # 启动消费者线程
        workers = []
        for _ in range(self.num_workers):
            t = threading.Thread(target=self._worker)
            t.start()
            workers.append(t)
        
        # 生产者：放入所有URL
        for url in self.urls:
            self.task_queue.put(url)
        
        # 放入结束信号（每个线程一个）
        for _ in range(self.num_workers):
            self.task_queue.put(None)
        
        # 等待所有任务完成
        self.task_queue.join()
        for t in workers:
            t.join()
        
        # 关闭缓冲区，确保数据落盘
        self.buffer.close()
        
        print(f"采集完成: 总计{self.stats['total']}, "
              f"成功{self.stats['success']}, 失败{self.stats['fail']}")

# 使用示例
pipeline = CollectPipeline(
    urls=['http://example.com/page/{}'.format(i) for i in range(1000)],
    output_file='data.jsonl',
    num_workers=10,
    buffer_size=200
)
pipeline.run()

4.6 性能调优建议

最后分享几个我在实战中总结的调优经验：

参数	建议值	说明
线程数	5-20	取决于目标网站和网络带宽
队列大小	200-1000	防止内存溢出，同时保证吞吐量
缓冲区大小	100-500条	平衡内存占用和磁盘IO
刷新间隔	5-30秒	防止数据丢失，不要太频繁

  核心要点回顾：
  IO密集型用多线程，CPU密集型用多进程
队列是生产者和消费者之间的缓冲带
数据缓冲区能大幅提升落盘效率
一定要加定时刷新，防止数据丢失
线程数不是越多越好，要测试找到最优值

这一章的内容就到这里。说白了，数据采集框架的核心就是并行、缓冲、落盘这三个环节。下一章我们会聊聊采集到的数据怎么清洗和预处理，那又是另一番天地了。

📚 课程目录

4、Python数据采集框架：多线程/多进程、队列管理与数据落盘

4.1 多线程 vs 多进程：到底选哪个？

4.2 实战：多线程采集框架

4.3 队列管理：生产者-消费者模式

4.4 数据缓冲：别让磁盘成为瓶颈

4.5 完整框架：把一切串起来

4.6 性能调优建议