如何用Python异步实现大文本中指定字符串的查找功能

设计说明 #

本地文件读取属于阻塞IO场景，asyncio没有原生高效的异步文件操作接口，aiohttp是HTTP请求专用库，完全不适用于本地文本处理，因此直接采用同步分块读取+匹配的方案，足够支持GB甚至TB级大文件处理，实现复杂度远低于异步方案。

模块实现 #

主模块 #

核心逻辑：

• 按设定的块大小（默认1000行）逐块读取文件
• 记录每个块的起始行偏移量，避免跨块匹配时行号计算错误
• 为避免关键词被块边界截断，每个块结尾预留最大关键词长度的字符重叠到下一个块
• 调用匹配器处理每个块，收集所有匹配结果后传给聚合器

def main(file_path: str, keywords: list[str], chunk_line_size: int = 1000):
    max_keyword_len = max(len(k) for k in keywords)
    all_match_result = {}
    current_line_offset = 0
    residual = ""  # 上一个块末尾预留的重叠内容
    residual_line_count = 0  # 重叠内容包含的行数

    with open(file_path, "r", encoding="utf-8") as f:
        while True:
            # 读取指定行数的块
            lines = []
            for _ in range(chunk_line_size):
                line = f.readline()
                if not line:
                    break
                lines.append(line)
            if not lines and not residual:
                break
            
            # 拼接上一个块的残留内容和当前块内容
            chunk_content = residual + "".join(lines)
            # 调用匹配器
            chunk_result = matcher(chunk_content, current_line_offset, keywords)
            # 合并结果到总结果
            for k, v in chunk_result.items():
                if k not in all_match_result:
                    all_match_result[k] = []
                all_match_result[k].extend(v)
            
            # 计算下一个块的行偏移和残留内容
            current_line_offset += len(lines) + residual_line_count
            # 截取当前块末尾的max_keyword_len长度作为残留
            if len(chunk_content) > max_keyword_len:
                residual = chunk_content[-max_keyword_len:]
                residual_line_count = residual.count("\n")
            else:
                residual = ""
                residual_line_count = 0
    
    # 调用聚合器输出
    aggregator(all_match_result)

匹配器 #

核心逻辑：

• 接收块内容、块起始行偏移、待匹配关键词集合
• 用正则表达式做多关键词匹配，自动处理关键词特殊字符转义
• 计算每个匹配结果的全局行偏移、字符偏移

import re
import bisect
from typing import List, Dict

def matcher(chunk_content: str, start_line_offset: int, keywords: list[str]) -> Dict[str, List[Dict]]:
    result = {k: [] for k in keywords}
    # 拼接正则模式，转义关键词特殊字符
    pattern = re.compile("|".join(re.escape(k) for k in keywords))
    # 提前计算每行的起始字符位置，用于匹配结果的行号计算
    line_start_pos = [0]
    for i, c in enumerate(chunk_content):
        if c == "\n":
            line_start_pos.append(i + 1)
    
    for match in pattern.finditer(chunk_content):
        match_str = match.group()
        match_start = match.start()
        # 计算匹配所在的行号（全局偏移）
        line_in_chunk = bisect.bisect_right(line_start_pos, match_start) - 1
        global_line_offset = start_line_offset + line_in_chunk
        # 计算匹配所在行的字符偏移
        char_in_line = match_start - line_start_pos[line_in_chunk]
        result[match_str].append({
            "lineOffset": global_line_offset,
            "charOffset": char_in_line
        })
    return result

聚合器 #

核心逻辑：

• 汇总所有块的匹配结果
• 按指定格式打印输出

def aggregator(all_result: Dict[str, List[Dict]]):
    for keyword, positions in all_result.items():
        if not positions:
            continue
        # 格式转换为要求的输出样式
        pos_str_list = [f"[lineOffset={p['lineOffset']}, charOffset={p['charOffset']}]" for p in positions]
        print(f"{keyword} --> [{', '.join(pos_str_list)}]")

调用示例 #

if __name__ == "__main__":
    TARGET_KEYWORDS = ["Timothy", "Jack", "Python"]
    main("big_text.txt", TARGET_KEYWORDS, chunk_line_size=1000)

可选优化项 #

• 若文件体积超过10GB、匹配关键词较多，可引入multiprocessing.Pool将匹配逻辑放到多进程执行，充分利用多核CPU性能
• 若关键词数量超过100个，可替换正则匹配为AC自动机算法，匹配效率提升10倍以上
• 可引入tqdm库增加读取进度显示，方便查看处理进度

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