You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何提升Python函数运行速度?scan_likes函数优化需求

优化大规模列表下的Python扫描函数性能

嘿,你的这个函数运行慢的核心问题其实很清晰:串行处理网络请求(每个请求都得等前一个完成)是最大的瓶颈,再加上大列表的遍历,自然耗时拉满。下面给你几个针对性的优化方案,能大幅提升运行速度:

1. 用并行请求替代串行for循环

网络请求属于IO密集型任务,并行处理能把等待响应的时间利用起来,直接砍掉大部分总耗时。这里推荐两种实现方式:

方案A:用ThreadPoolExecutor快速改造

这个方案改动最小,不需要换HTTP库,适合快速上手:

import requests
from concurrent.futures import ThreadPoolExecutor

def process_single_element(element):
    match_count = 0
    try:
        # 注意:原代码里的`header`应该是`headers`(requests的正确参数名),这个小错误可能也会导致异常
        response = requests.get('https://booksmania.com', headers=element)
        # 优化文本检查:字符串内置的count方法是C实现的,比Python循环快N倍
        match_count = response.text.count('ok')
    # 尽量捕获具体异常,避免隐藏未知问题
    except requests.exceptions.RequestException:
        pass
    return (element, match_count)

def scan_likes(my_list):
    result_counts = {}
    # 根据你的机器性能调整max_workers,一般10-50之间合适,太高反而会因为线程切换耗资源
    with ThreadPoolExecutor(max_workers=20) as executor:
        # 批量提交所有任务,并行执行
        for elem, cnt in executor.map(process_single_element, my_list):
            result_counts[elem] = cnt
    return result_counts

这里的关键优化:

  • 把单个element的处理逻辑抽成独立函数,交给线程池并行执行
  • 用response.text.count('ok')替代逐行遍历,大幅提升文本检查效率

方案B:用异步HTTP库(推荐大规模场景)

如果你的my_list规模极大(比如上万级别的元素),用aiohttp异步请求会比线程池更节省资源,性能也更优:

import aiohttp
import asyncio

async def process_single_element(session, element):
    match_count = 0
    try:
        async with session.get('https://booksmania.com', headers=element) as response:
            # 要是需要统计次数,直接获取文本后用count
            text = await response.text()
            match_count = text.count('ok')
            # 要是只需要判断是否存在,用`if 'ok' in text`就够了
    except aiohttp.ClientError:
        pass
    return (element, match_count)

async def scan_likes(my_list):
    result_counts = {}
    # 复用ClientSession,减少连接建立的开销
    async with aiohttp.ClientSession() as session:
        # 创建所有异步任务
        tasks = [process_single_element(session, elem) for elem in my_list]
        # 并行执行所有任务并收集结果
        results = await asyncio.gather(*tasks)
        for elem, cnt in results:
            result_counts[elem] = cnt
    return result_counts

# 调用方式(注意异步函数需要用asyncio.run触发)
# final_counts = asyncio.run(scan_likes(your_large_list))

异步方案的优势在于:不需要创建大量线程,用单线程就能处理成百上千的并发请求,资源占用更低,适合超大规模的请求场景。

2. 其他细节优化

  • 去重处理:如果my_list里有重复的element,先去重再处理,能减少重复的网络请求:my_list = list(dict.fromkeys(my_list))(保持原顺序的去重方式)
  • 避免无差别异常捕获:原代码里的except Exception:会捕获所有异常,包括语法错误、逻辑错误这类不该忽略的问题,建议只捕获和网络请求相关的具体异常
  • 减少不必要的字典初始化:原代码里先给count[element] = 0,之后再累加,其实可以直接在处理函数里返回结果,最后统一赋值给字典,少一步操作

内容的提问来源于stack exchange,提问作者R Artur

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.05.08 07:12:52