使用pytube多线程获取YouTube流无性能提升问题排查

算力阿峰

2026-8-2

问题描述

使用pytube库批量获取YouTube视频流信息时，分别尝试了线性循环、concurrent.futures线程池、multiprocessing进程池三种方式，但三者耗时均约130秒，未实现预期的性能提升。

核心代码如下：

def _get_streams(self, yt: pytube.YouTube):
    return yt.streams

@print_execution_time
def get_streams_futures(self):
    with concurrent.futures.ThreadPoolExecutor() as executor:
        future_results = [executor.submit(self._get_streams, yt) for yt in self._create_yt_obj()]
        result = [future.result() for future in concurrent.futures.as_completed(future_results)]
    return result

@print_execution_time
def get_streams_linear(self):
    return [yt.streams for yt in self._create_yt_obj()]

@print_execution_time
def get_streams_multiprocessing(self):
    with multiprocessing.Pool() as pool:
        result = pool.map(self._get_streams, self._create_yt_obj())

    return result

@print_execution_time是用于统计函数运行时间的装饰器，运行无报错，需分析现象原因并给出性能提升方案。

原因分析

耗时核心在_create_yt_obj()而非_get_streams()
当前逻辑是先一次性创建所有pytube.YouTube对象，再调用yt.streams获取流信息。但pytube.YouTube初始化时已经完成了大部分网络请求（拉取视频元数据、解析页面），这才是真正的耗时环节。后续yt.streams只是读取本地缓存的数据，几乎不占用时间。所以线程池/进程池只是在处理无耗时的本地操作，自然无法提升性能。
线程池未作用于IO密集型环节
网络请求属于IO密集型任务，线程池本应发挥并行优势，但你把最耗时的网络请求放在了创建YouTube对象阶段，线程池根本没参与到核心耗时操作中。此外，pytube内部可能存在同步锁，进一步限制了线程并行效率。
进程池额外开销抵消收益
进程池创建进程、序列化/反序列化YouTube对象都有额外开销，而_get_streams本身几乎不耗时，这些开销反而让整体耗时和线性方式持平。

性能提升方案

将网络请求逻辑移入并行任务
不要提前创建YouTube对象，而是将视频URL传入任务函数，在任务内部完成初始化和流信息获取，让线程池/进程池真正并行处理网络请求。修改示例：

def _fetch_streams_from_url(self, video_url: str):
    yt = pytube.YouTube(video_url)
    return yt.streams

@print_execution_time
def get_streams_futures_optimized(self):
    # 假设self.video_urls是存储所有视频URL的列表
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        future_results = [executor.submit(self._fetch_streams_from_url, url) for url in self.video_urls]
        result = [future.result() for future in concurrent.futures.as_completed(future_results)]
    return result

优先使用线程池
获取视频流是IO密集型任务，线程池的开销远低于进程池，更适合这类场景。可根据网络情况调整max_workers参数（建议10-20），避免并发过高触发YouTube的请求限制。

添加请求限流与重试
YouTube会对高频请求做拦截，过于激进的并发可能导致请求失败反而增加耗时。可以在任务函数中添加随机延迟和重试逻辑：

import time
import random
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def _fetch_streams_from_url(self, video_url: str):
    time.sleep(random.uniform(0.5, 1.5))  # 添加随机延迟分散请求
    yt = pytube.YouTube(video_url)
    return yt.streams

尝试异步版本pytube（可选）
官方pytube是同步实现，可使用第三方维护的异步分支（如async-pytube）结合asyncio实现异步IO，进一步提升并行效率。

内容的提问来源于stack exchange，提问作者Davi A. Sampaio