使用asyncio调用TikTokApi多会话超时问题求助

问题分析 #

你遇到的超时本质是一次性创建过多浏览器会话导致的双重问题：

• 50个Chromium实例同时启动、加载TikTok页面，会占用大量CPU/内存，导致页面加载速度骤降，超出默认30秒超时阈值
• TikTok反爬机制会对短时间内的大量并发连接限流，直接触发超时

可行解决方案 #

1. 延长会话创建的超时时间 #

create_sessions默认超时为30000ms（30秒），针对50+会话场景，直接调大超时阈值：

await api.create_sessions(
    ms_tokens=[ms_token], 
    num_sessions=50, 
    sleep_after=3,
    timeout=60000  # 改为60秒，可根据实际运行情况再调整
)

2. 增大会话创建间隔，避免并发过载 #

调大sleep_after参数，给每个会话的启动和页面加载留足缓冲时间，同时降低被TikTok反爬检测的概率：

await api.create_sessions(
    ms_tokens=[ms_token], 
    num_sessions=50, 
    sleep_after=5,  # 从3秒改为5秒或更长
    timeout=60000
)

3. 分批创建会话，而非一次性启动所有 #

如果直接调参仍超时，可把50个会话分成多批创建，每批完成后短暂休眠：

async def fetch():
    async with TikTokApi() as api:
        # 分2批创建，每批25个
        batch_size = 25
        total_sessions = 50
        for i in range(0, total_sessions, batch_size):
            await api.create_sessions(
                ms_tokens=[ms_token], 
                num_sessions=batch_size, 
                sleep_after=3,
                timeout=60000
            )
            await asyncio.sleep(2)  # 批与批之间休眠2秒
        
        video_list = []
        async for video in api.trending.videos(count=1):
            video_list.append(video.as_dict)

        return video_list

4. 禁用浏览器不必要资源，降低内存占用 #

通过browser_args禁用图片、非必要JS等资源，减少每个浏览器实例的内存消耗，让更多会话稳定运行：

await api.create_sessions(
    ms_tokens=[ms_token], 
    num_sessions=50, 
    sleep_after=3,
    timeout=60000,
    browser_args=[
        "--blink-settings=imagesEnabled=false",
        "--disable-css",
        "--no-sandbox",
        "--disable-dev-shm-usage"
    ]
)

5. 使用多个ms_token分摊负载 #

单个ms_token存在并发请求限制，准备多个有效ms_token分配给不同会话，降低单个账号的压力：

# 示例：从环境变量读取多个ms_token
ms_tokens = [os.environ.get(f"ms_token_{i}") for i in range(3)]

await api.create_sessions(
    ms_tokens=ms_tokens, 
    num_sessions=50, 
    sleep_after=3,
    timeout=60000
)

额外优化提示 #

爬取视频时用asyncio.Semaphore控制并发请求数，避免一次性发起过多请求触发限流：

async def fetch():
    async with TikTokApi() as api:
        await api.create_sessions(ms_tokens=[ms_token], num_sessions=50, sleep_after=3, timeout=60000)
        semaphore = asyncio.Semaphore(10)  # 限制同时处理10个视频
        video_list = []
        
        async def safe_fetch_video(video):
            async with semaphore:
                video_list.append(video.as_dict)
        
        tasks = []
        async for video in api.trending.videos(count=1):
            tasks.append(asyncio.create_task(safe_fetch_video(video)))
        
        await asyncio.gather(*tasks)
        return video_list

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