Python多线程处理HTML解析性能下降，如何优化？

我正在学习Python并发编程，写了一个简单的HTML解析器——从本地目录读取HTML文件，解析指定字段后保存为JSON到另一个目录。原本预期多线程能提升性能，但测试发现不管处理50、200、1000甚至30000个文件，多线程速度都比单线程慢。比如处理1000个文件时，单线程耗时约2.9秒，多线程要4秒。

试过threading模块和concurrent.futures.ThreadPoolExecutor，结果一致。我知道GIL，也明白IO密集型任务适合多线程，但实际结果相反。

我的代码如下：

import json
import re
import time
from pathlib import Path
import threading


def get_json_data(body: str) -> re.Match[str] or None:
    return re.search(
        r'(?<=json_data&quot;>)(.*?)(?=</script>)', body
    )


def parse_html_file(file_path: Path) -> dict:
    with open(file_path, "r") as file:
        html_content = file.read()
        match = get_json_data(html_content)
        if not match:
            return {}

        next_data = match.group(1)
        json_data = json.loads(next_data)

        data1 = json_data.get("data1")
        data2 = json_data.get("data2")
        data3 = json_data.get("data3")
        data4 = json_data.get("data4")
        data5 = json_data.get("data5")

        parsed_fields = {
            "data1": data1,
            "data2": data2,
            "data3": data3,
            "data4": data4,
            "data5": data5
        }

        return parsed_fields


def save_parsed_fields(file_path: Path, parsed_fields: dict, output_dir: Path) -> None:
    output_filename = f"parsed_{file_path.stem}.json"
    output_path = output_dir / output_filename

    with open(output_path, "w") as output_file:
        json.dump(parsed_fields, output_file)

    print(f"Parsed {file_path.name} and saved the results to {output_path}")


def process_html_file(file_path: Path, parsed_dir: Path) -> None:
    parsed_fields = parse_html_file(file_path)
    save_parsed_fields(file_path, parsed_fields, parsed_dir)


def process_html_files(source_dir: Path, parsed_dir: Path) -> None:
    parsed_dir.mkdir(parents=True, exist_ok=True)

    threads = []
    for file_path in source_dir.glob("*.html"):
        thread = threading.Thread(target=process_html_file, args=(file_path, parsed_dir))
        thread.start()
        threads.append(thread)

    # Wait for all threads to finish
    for thread in threads:
        thread.join()


def main():
    base_path = "/home/my_pc/data"
    source_dir = Path(f"{base_path}/html_sample")
    parsed_dir = Path(f"{base_path}/parsed_sample")

    start_time = time.time()

    process_html_files(source_dir, parsed_dir)

    end_time = time.time()
    duration = end_time - start_time
    print(f"Application took {duration:.2f} seconds to complete.")


if __name__ == "__main__":
    main()

使用ThreadPoolExecutor的代码片段：

with ThreadPoolExecutor(max_workers=max_workers) as executor:
    # Iterate over the HTML files in the source directory
    for file_path in source_dir.glob("*.html"):
        executor.submit(process_html_file, file_path, parsed_dir)

请问我的代码有问题吗？不使用asyncio的前提下，怎么优化多线程实现？

为什么多线程反而变慢？ #

1. 线程创建/销毁开销过大：当前实现为每个文件创建一个线程，文件数量多时，线程创建、调度的开销远超过IO等待节省的时间，拖慢整体速度。
2. GIL的隐性影响：json.loads和正则匹配属于CPU密集操作，会占用GIL，导致线程间频繁切换，增加额外开销。
3. 磁盘IO瓶颈：本地磁盘带宽有限，多线程同时读写会引发磁盘争用，尤其是机械硬盘，随机读写的并发性能极差。
4. print语句的锁竞争：print是线程安全操作，内部会加锁，大量线程同时调用会导致锁竞争，增加耗时。

优化方案 #

1. 限制线程池大小，避免无限制创建线程

用ThreadPoolExecutor并设置合理的max_workers值，对于磁盘IO密集任务，建议设置为CPU核心数的2-4倍（机械硬盘设4-8，SSD设8-16），避免线程过多导致调度和IO争用。

修改后的process_html_files函数：

from concurrent.futures import ThreadPoolExecutor

def process_html_files(source_dir: Path, parsed_dir: Path) -> None:
    parsed_dir.mkdir(parents=True, exist_ok=True)
    # 根据磁盘类型调整max_workers
    max_workers = 8
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        for file_path in source_dir.glob("*.html"):
            executor.submit(process_html_file, file_path, parsed_dir)

2. 移除或批量处理print语句

大量print会带来锁竞争，建议去掉打印，或者收集结果后批量打印：

# 修改save_parsed_fields，返回结果而非直接打印
def save_parsed_fields(file_path: Path, parsed_fields: dict, output_dir: Path) -> str:
    output_filename = f"parsed_{file_path.stem}.json"
    output_path = output_dir / output_filename

    with open(output_path, "w") as output_file:
        json.dump(parsed_fields, output_file)

    return f"Parsed {file_path.name} and saved the results to {output_path}"

# 修改process_html_file，返回保存结果
def process_html_file(file_path: Path, parsed_dir: Path) -> str:
    parsed_fields = parse_html_file(file_path)
    return save_parsed_fields(file_path, parsed_fields, parsed_dir)

# 批量处理结果打印
def process_html_files(source_dir: Path, parsed_dir: Path) -> None:
    parsed_dir.mkdir(parents=True, exist_ok=True)
    max_workers = 8
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(process_html_file, file_path, parsed_dir) 
                   for file_path in source_dir.glob("*.html")]
        for future in futures:
            results.append(future.result())
    # 批量打印
    for res in results:
        print(res)

3. 优化CPU密集部分的代码

• 预编译正则表达式，避免每次调用都重新编译：

# 预编译正则，放在函数外部
JSON_DATA_PATTERN = re.compile(r'(?<=json_data&quot;>)(.*?)(?=</script>)')

def get_json_data(body: str) -> re.Match[str] or None:
    return JSON_DATA_PATTERN.search(body)

• 用更快的JSON库（如ujson）替代标准库json，解析速度可提升数倍。

4. 调整文件读写方式

如果是机械硬盘，可按文件系统的顺序处理文件（排序后遍历），减少磁盘寻道时间；SSD则无需此操作，但也应避免无意义的随机读写。

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