Python3使用multiprocessing.pool处理文件行时共享计数器失效问题求助

• 你此前查到的「创建线程时拷贝所有数据」是多进程的特性，多线程共享当前进程的内存空间，Python3中线程修改全局变量本身对主线程可见，你遇到的计数器不工作问题核心是3点：
  1. 原代码缺失sys、time模块导入，且未实现getElapsedTime工具函数，子线程执行时直接抛出异常，但apply_async默认不会主动抛出子线程错误，导致计数逻辑完全没有执行
  2. 多线程并发修改全局计数器、打印进度时没有加锁，会出现计数丢失、打印内容错乱的问题
  3. 全局匹配标记found_matched_line的修改没有同步机制，主线程可能无法及时感知到匹配状态，不会及时停止文件读取

完全保留你原有的异步处理逻辑，仅新增线程锁和补充缺失代码即可，额外性能损耗极低，符合你的性能要求：

• 新增全局线程锁，所有修改全局变量、打印进度的操作都加锁，避免竞态问题
• 补充缺失的导入和getElapsedTime函数实现
• 给apply_async增加错误回调，方便后续排查子线程异常

from multiprocessing.pool import ThreadPool as Pool
from datetime import datetime
from threading import Lock
import sys
import time

pool_size = 10  # 并行线程数
progress_output_interval = 20  # 每处理多少行打印一次进度

# 全局状态
found_matched_line = False
found_matched_line_val = ''
count_lines_tested = 0
count_batches_tested = 0
global_lock = Lock()  # 新增：全局线程锁

start_time = datetime.now()

# 新增：缺失的时间格式化函数
def getElapsedTime(start, end):
    delta = end - start
    hours = delta.seconds // 3600
    minutes = (delta.seconds % 3600) // 60
    seconds = delta.seconds % 60
    return f"{hours:02d}:{minutes:02d}:{seconds:02d}"

def process_file (file_name, process_function, param1, param2, param3, param4):
    pool = Pool(pool_size)
    global found_matched_line
    global count_batches_tested 
    batch = ['0'] * pool_size
    curr_batch_size = 0
    end_of_file = False

    # 打开文件，非ASCII文件可自行指定encoding参数，比如encoding='utf-8'
    f = open(file_name, 'r')

    while not end_of_file and not found_matched_line: 
        curr_batch_size = 0
        # 读取批次数据
        for i in range(pool_size):
            curr_line = f.readline()
            if not curr_line: 
                end_of_file = True
            else: 
                curr_line = curr_line.rstrip()
                batch[i] = curr_line
                curr_batch_size += 1

        # 提交异步任务
        for j in range(curr_batch_size):
            curr_line2 = batch[j]
            if not curr_line2: 
                print('Info - skipping blank line')
            else: 
                # 新增error_callback，子线程异常时会打印错误
                if process_function == 'process_line_function1':
                    pool.apply_async(process_line_function1, (curr_line2, param1, param2, param3, param4), error_callback=lambda e: print(f"子线程执行错误: {e}"))
                elif process_function == 'process_line_function2':
                    pool.apply_async(process_line_function2, (curr_line2, param1, param2, param3), error_callback=lambda e: print(f"子线程执行错误: {e}"))
                else: 
                    print('Error: unexpected process_function '+str(process_function))

        if curr_batch_size > 0:
            count_batches_tested += 1
            
    pool.close()
    pool.join()
    f.close()
    return


def process_line_function1 (line, param1, param2, param3, param4):
    match_found = False
    # 此处为你的匹配逻辑，测试时可自行修改触发条件，比如 match_found = line == 'test'
    time.sleep(.3)

    if match_found: 
        # 修改全局状态前加锁
        with global_lock:
            global found_matched_line
            found_matched_line = True
            global found_matched_line_val
            found_matched_line_val = line
            print('Found a match: '+found_matched_line_val)

    # 进度更新逻辑加锁
    with global_lock:
        global count_lines_tested
        count_lines_tested += 1
        now_time = datetime.now()
        if (count_lines_tested % progress_output_interval) == 0:
            overall_elapsed_time = getElapsedTime(start_time, now_time)
            overall_elapsed_time_seconds_decimal = (now_time - start_time).total_seconds()
            overall_rate = '{:.2f}'.format(count_lines_tested/overall_elapsed_time_seconds_decimal)
            print(str(count_lines_tested)+' tested'+', elapsed: '+overall_elapsed_time+', rate: '+overall_rate+' lines/sec')
    return


def process_line_function2 (line, param1, param2, param3):
    # 逻辑同process_line_function1，按需实现即可
    match_found = False
    time.sleep(.3)

    if match_found: 
        with global_lock:
            global found_matched_line
            found_matched_line = True
            global found_matched_line_val
            found_matched_line_val = line
            print('Found a match: '+found_matched_line_val)

    with global_lock:
        global count_lines_tested
        count_lines_tested += 1
        now_time = datetime.now()
        if (count_lines_tested % progress_output_interval) == 0:
            overall_elapsed_time = getElapsedTime(start_time, now_time)
            overall_elapsed_time_seconds_decimal = (now_time - start_time).total_seconds()
            overall_rate = '{:.2f}'.format(count_lines_tested/overall_elapsed_time_seconds_decimal)
            print(str(count_lines_tested)+' tested'+', elapsed: '+overall_elapsed_time+', rate: '+overall_rate+' lines/sec')
    return


if __name__ == '__main__':
    file_name = sys.argv[1]
    process_line_function = ''
    # 此处为你的处理函数选择逻辑，按需修改即可
    if True:
        process_line_function = 'process_line_function1'
    else:
        process_line_function = 'process_line_function2'

    param1 = 'blah'
    param2 = 'blah2'
    param3 = 'blah3'
    param4 = 'blah4'

    process_file(file_name, process_line_function, param1, param2, param3, param4)

    # 最终统计
    end_time = datetime.now()
    elapsed_time_seconds_decimal = (end_time - start_time).total_seconds()
    final_rate = "{:.2f}".format(count_lines_tested/elapsed_time_seconds_decimal)
    print('Started at: '+start_time.strftime("%Y-%m-%d %H:%M:%S"))
    print('Finished at: '+end_time.strftime("%Y-%m-%d %H:%M:%S"))
    print(str(count_lines_tested)+' lines tested (in '+str(count_batches_tested)+' batches)')
    print('Rate: '+final_rate+' lines/second')
    if found_matched_line: 
        print('MATCH WAS FOUND! --- '+found_matched_line_val)
    else: 
        print('Match was not found')

    sys.exit(1)

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