如何进一步优化Python多进程哈希校验脚本以缩短处理时间？

看起来你已经做了不错的多进程优化，但结合你的VPS配置（4vCPU）和当前的瓶颈点（bcrypt校验的计算开销、进程间通信、进程数与核心数不匹配），咱们可以从以下几个方向进一步压缩处理时间：

1. 对齐进程数与CPU核心数，减少上下文切换 #

你的VPS只有4个vCPU，但当前设置了6个工作进程——超过核心数的进程会导致操作系统频繁切换进程上下文，反而拖慢整体速度。建议把进程数和分块数调整为和CPU核心数一致，甚至可以动态获取核心数：

# 动态获取CPU核心数，fallback到4
MAX_WORKERS = os.cpu_count() or 4
MAX_CHUNKS = MAX_WORKERS

这样每个进程能更高效地独占CPU时间片，避免不必要的调度开销。

2. 优化进程间通信（IPC）的开销 #

你当前用Manager.dict来做全局的found_flag，但字典类型的共享对象IPC开销比简单的布尔值大很多。换成multiprocessing.Value（轻量级的共享布尔值）能显著减少进程间的通信成本：

# 先导入Value
from multiprocessing import Manager, Pool, Value

# 修改process_chunk函数的flag检查逻辑
def process_chunk(chunk, stored_hash, found_flag):
    for document in chunk:
        # 直接访问共享Value的value属性，比dict更高效
        if found_flag.value:
            break
        if verify_password(stored_hash, document):
            found_flag.value = True
            return document
    return None

# 在main函数里替换found_flag的定义
def main(stored_hash):
    documents = get_documents()
    start_time = time.time()

    with Manager() as manager:
        # 'b'代表C语言的布尔类型，初始值为False
        found_flag = Value('b', False)
        chunks = chunk_documents(documents, MAX_CHUNKS)

        with Pool(processes=MAX_WORKERS) as pool:
            results = [
                pool.apply_async(process_chunk, args=(chunk, stored_hash, found_flag))
                for chunk in chunks
            ]
            for result in results:
                output = result.get()
                if output is not None:
                    pool.terminate()
                    end_time = time.time()
                    return output

    end_time = time.time()
    return ''

另外，还可以减少检查found_flag的频率——比如每检查10个文档再判断一次是否已经找到匹配，进一步降低IPC次数（毕竟每次跨进程访问共享变量都有开销）：

def process_chunk(chunk, stored_hash, found_flag):
    for idx, document in enumerate(chunk):
        # 每10条文档检查一次flag，平衡响应速度和IPC开销
        if idx % 10 == 0 and found_flag.value:
            break
        if verify_password(stored_hash, document):
            found_flag.value = True
            return document
    return None

3. 预处理文档号，避免重复编码 #

当前verify_password里每次都要把document编码成utf-8，你可以提前在get_documents里把所有文档号编码好，减少重复计算：

def get_documents():
    connection = mysql.connector.connect(
        host=DB_HOST,
        database=DB_DATABASE,
        user=DB_USERNAME,
        password=DB_PASSWORD,
        port=DB_PORT
    )
    cursor = connection.cursor()
    query = "SELECT nro_document FROM family_and_friends WHERE deleted_at IS NULL ORDER BY id ASC;"
    cursor.execute(query)
    # 提前把所有文档号编码成bytes，避免在每个进程里重复encode
    documents = [row[0].encode('utf-8') for row in cursor.fetchall()]
    cursor.close()
    connection.close()
    return documents

# 同时修改verify_password，去掉重复的encode步骤
def verify_password(stored_hash, password_to_verify):
    # 现在password_to_verify已经是bytes类型了
    return bcrypt.checkpw(password_to_verify, stored_hash)

这样每个文档号只需要编码一次，而不是在每个进程里重复执行编码操作。

4. 优化bcrypt的校验效率 #

bcrypt的12轮确实安全，但计算成本是指数级增长的——12轮的计算量是10轮的4倍。如果业务场景允许（比如对哈希强度要求不是极端严格），把工作因子降低到10轮，校验速度会提升约40%。如果不能降低轮数，确保你安装的是官方的bcrypt包（基于C扩展实现），而不是纯Python的版本，后者速度会慢很多。

5. 提前终止进程的优化 #

当前代码要等result.get()才会终止Pool，但其实可以用回调函数，一旦有进程返回匹配结果，立刻终止所有进程，减少不必要的计算：

def main(stored_hash):
    documents = get_documents()
    start_time = time.time()
    found_result = None

    with Manager() as manager:
        found_flag = Value('b', False)
        chunks = chunk_documents(documents, MAX_CHUNKS)

        def handle_result(output):
            nonlocal found_result
            if output is not None:
                found_result = output
                pool.terminate()  # 找到结果立刻终止所有进程，停止无用计算

        with Pool(processes=MAX_WORKERS) as pool:
            for chunk in chunks:
                pool.apply_async(
                    process_chunk, 
                    args=(chunk, stored_hash, found_flag), 
                    callback=handle_result
                )
            pool.close()
            pool.join()  # 等待所有进程结束（或者被终止）

        if found_result is not None:
            end_time = time.time()
            return found_result

    end_time = time.time()
    return ''

这样不需要等待所有进程的结果返回，一旦找到匹配就立刻终止所有进程，节省后续的无效计算时间。

6. 数据库连接的小优化 #

当前的get_documents每次都创建新的数据库连接，对于580条记录来说影响不大，但如果后续记录增多，可以考虑用数据库连接池（比如mysql-connector-python的pooling模块，或者pymysql配合DBUtils连接池），减少连接建立的开销。另外，mysql.connector的性能略逊于pymysql，如果允许的话，可以替换成pymysql试试。

结合你的580条记录，这些优化应该能把40秒的处理时间大幅压缩——比如对齐进程数+预处理编码+优化IPC，预计能减少30%-50%的时间，如果再调整bcrypt轮数，能进一步提速。

备注：内容来源于stack exchange，提问作者beker32