1. 问题背景与挑战
在处理大规模数据集时，例如需要在一个包含数万条记录的json列表中（json_list）查找并匹配另一个包含数千个标记（marking）的列表中的元素，性能往往成为一个关键挑战。具体场景是，json_list中的每个字典包含一个code字段，我们需要将marking列表中的每个字符串与json_list中元素的code字段进行相似度匹配。匹配规则是使用difflib.sequencematcher计算相似度，当相似度为1（完全匹配）或介于0.98到0.99之间时，认为匹配成功。
原始的单线程或简单的多线程（threading）实现，在数据量庞大时（json_list超过23,000条，marking超过3,000条），可能需要20分钟甚至更长时间才能完成。这主要是因为Python的全局解释器锁（GIL）限制了多线程在CPU密集型任务上的并行执行能力。即使创建了多个线程，它们也无法同时在多个CPU核心上运行Python字节码，导致性能提升不明显。
2. 多进程（Multiprocessing）的解决方案
为了克服GIL的限制，Python提供了multiprocessing模块，它允许创建独立的进程，每个进程都有自己的Python解释器和内存空间。这意味着不同的进程可以在不同的CPU核心上真正并行执行CPU密集型任务，从而显著提高性能。
本教程将展示如何利用multiprocessing库来优化上述数据匹配和筛选过程。
2.1 核心组件介绍
- multiprocessing.Process: 用于创建和管理新的进程。
- multiprocessing.Manager: 用于创建可以在不同进程之间共享的数据结构（如列表、字典等）。这是解决进程间通信和数据共享的关键，因为普通Python对象在进程间默认不共享。
- difflib.SequenceMatcher: 用于计算两个序列（字符串）的相似度。
2.2 匹配逻辑函数 find_marking
这个函数负责执行单个标记与JSON数据项的匹配逻辑。它保持不变，因为它是一个纯计算函数，不涉及并发问题。
立即学习“Python免费学习笔记（深入）”；
from difflib import SequenceMatcher
def find_marking(x: str, y: dict) -> dict | None:
"""
比较标记字符串x与JSON数据项y中的'code'字段的相似度。
如果相似度为1或在0.98到0.99之间，则返回y，否则返回None。
"""
text_match = SequenceMatcher(None, x, y.get('code', '')).ratio()
if text_match == 1 or (0.98 <= text_match < 0.99):
return y
return None注意： 确保y字典中包含'code'键，否则y.get('code', '')可以提供一个默认值，避免KeyError。
2.3 多进程筛选主函数 eliminate_marking
这个函数是整个解决方案的核心，它协调多个进程来并行处理匹配任务。
import math
from multiprocessing import Process, Manager
def eliminate_marking(marking_list: list[str], json_list: list[dict]) -> tuple[list[str], list[dict]]:
"""
使用多进程并行地从json_list中匹配和筛选marking_list中的标记。
Args:
marking_list: 待匹配的标记字符串列表。
json_list: 包含'code'字段的JSON字典列表。
Returns:
一个元组，包含两个列表：
- result_mark: 成功匹配的标记列表。
- result: 成功匹配的JSON数据项列表。
"""
# 1. 初始化Manager和共享数据结构
# Manager用于创建可在进程间共享的列表，以收集结果。
manager = Manager()
result_mark = manager.list() # 共享列表，用于存储成功匹配的标记
result = manager.list() # 共享列表，用于存储成功匹配的JSON数据项
def __process_eliminate(sub_marking_list: list[str], data_scrap: list[dict],
shared_result_mark: Manager.list, shared_result: Manager.list):
"""
每个进程执行的任务函数。
它遍历分配给它的标记子列表，并尝试在data_scrap中找到匹配项。
"""
# data_scrap是json_list的一个副本，每个进程独立操作。
# 注意：这里的data_scrap是json_list的浅拷贝，对其内部字典的修改会影响原始字典
# 但对其列表结构（如remove操作）的修改仅影响当前进程的副本。
# 鉴于我们的目标是收集匹配项，这种拷贝方式是安全的。
for marking_item in sub_marking_list: # 遍历当前进程负责的标记子列表
for data in data_scrap: # 遍历json_list的副本
result_data = find_marking(marking_item, data)
if result_data:
# 找到匹配项后，将其添加到共享列表中
shared_result_mark.append(marking_item)
shared_result.append(result_data)
# 这里的remove操作只影响当前进程的data_scrap副本，
# 并不影响其他进程的副本或原始json_list。
# 如果目标是真正从原始json_list中移除，需要更复杂的同步机制。
# 在当前场景下，我们主要关注收集匹配结果。
# data_scrap.remove(data)
# 如果一个标记只需要匹配一次，可以在找到后跳出内层循环
break # 一个marking_item找到一个匹配后就跳出，避免重复匹配
# 2. 任务分块与进程创建
processes = []
chunk_size = 100 # 每个进程处理的marking_list块的大小
# 计算需要创建的进程数量
# 这里将marking_list分成块，每个进程处理一个或多个块。
# 另一种常见策略是基于CPU核心数创建进程。
num_chunks = math.ceil(len(marking_list) / chunk_size)
for i in range(num_chunks):
start_idx = i * chunk_size
end_idx = min((i + 1) * chunk_size, len(marking_list))
sub_marking_list = marking_list[start_idx:end_idx]
if not sub_marking_list:
continue # 避免创建空任务的进程
p = Process(
target=__process_eliminate,
# args参数传递给目标函数。
# json_list[:] 创建了一个json_list的浅拷贝，确保每个进程有独立的副本。
args=(sub_marking_list, json_list[:], result_mark, result)
)
processes.append(p)
p.start() # 启动进程
# 3. 等待所有进程完成
for p in processes:
p.join() # 阻塞主进程，直到当前进程执行完毕
# 4. 关闭Manager并返回结果
manager.shutdown() # 在所有进程完成后关闭Manager
return list(result_mark), list(result) # 将Manager.list转换为普通list返回
2.4 完整示例代码
为了方便测试，我们创建一些模拟数据：
import math
import time
import random
import string
from difflib import SequenceMatcher
from multiprocessing import Process, Manager
# 模拟数据
def generate_fake_data(num_json, num_marking):
json_list = []
for i in range(num_json):
code_val = ''.join(random.choices(string.digits, k=6))
json_list.append({
"code": code_val,
"phone_number": f"1{random.randint(1000000000, 9999999999)}",
"email": f"user{i}@example.com",
"address": f"address_fake_{i}",
"note": f"note dummy {i}"
})
marking = []
# 确保有一些匹配项
for i in range(num_marking // 2):
# 从json_list中随机取一个code作为marking
marking.append(random.choice(json_list)['code'])
# 添加一些不匹配的marking
for i in range(num_marking // 2, num_marking):
marking.append(''.join(random.choices(string.ascii_letters + string.digits, k=random.randint(5, 8))))
random.shuffle(marking) # 打乱顺序
return json_list, marking
# 假设的 find_marking 函数
def find_marking(x: str, y: dict) -> dict | None:
text_match = SequenceMatcher(None, x, y.get('code', '')).ratio()
if text_match == 1 or (0.98 <= text_match < 0.99):
return y
return None
# 假设的 eliminate_marking 函数（与上面定义的一致）
def eliminate_marking(marking_list: list[str], json_list: list[dict]) -> tuple[list[str], list[dict]]:
manager = Manager()
result_mark = manager.list()
result = manager.list()
def __process_eliminate(sub_marking_list: list[str], data_scrap: list[dict],
shared_result_mark: Manager.list, shared_result: Manager.list):
for marking_item in sub_marking_list:
for data in data_scrap:
result_data = find_marking(marking_item, data)
if result_data:
shared_result_mark.append(marking_item)
shared_result.append(result_data)
break # 一个marking_item找到一个匹配后就跳出
processes = []
# 这里的chunk_size可以根据实际CPU核心数和任务复杂度进行调整
# 较小的chunk_size可能导致更多的进程创建和管理开销
# 较大的chunk_size可能导致部分核心利用率不足
chunk_size = 50 # 调整为50，以创建更多进程进行测试，更细粒度的任务分配
# 优化：根据CPU核心数来决定进程数量，而不是简单地按chunk_size分块
# 理想情况下，进程数不应超过CPU核心数
# num_processes = os.cpu_count() or 1
# marking_per_process = math.ceil(len(marking_list) / num_processes)
#
# for i in range(num_processes):
# start_idx = i * marking_per_process
# end_idx = min((i + 1) * marking_per_process, len(marking_list))
# sub_marking_list = marking_list[start_idx:end_idx]
# ...
# 当前实现是按chunk_size分块
num_chunks = math.ceil(len(marking_list) / chunk_size)
for i in range(num_chunks):
start_idx = i * chunk_size
end_idx = min((i + 1) * chunk_size, len(marking_list))
sub_marking_list = marking_list[start_idx:end_idx]
if not sub_marking_list:
continue
p = Process(
target=__process_eliminate,
args=(sub_marking_list, json_list[:], result_mark, result)
)
processes.append(p)
p.start()
for p in processes:
p.join()
manager.shutdown()
return list(result_mark), list(result)
if __name__ == "__main__":
# 生成模拟数据
NUM_JSON = 23000
NUM_MARKING = 3000
print(f"生成 {NUM_JSON} 条JSON数据和 {NUM_MARKING} 条标记数据...")
test_json_list, test_marking_list = generate_fake_data(NUM_JSON, NUM_MARKING)
print("数据生成完毕。")
start_time = time.time()
eliminated_markings, eliminated_data = eliminate_marking(test_marking_list, test_json_list)
end_time = time.time()
print(f"\n多进程处理完成。")
print(f"总耗时: {end_time - start_time:.2f} 秒")
print(f"找到 {len(eliminated_markings)} 个匹配标记。")
print(f"找到 {len(eliminated_data)} 条匹配数据。")
# 验证部分结果
if eliminated_markings:
print(f"部分匹配标记示例: {eliminated_markings[:5]}")
if eliminated_data:
print(f"部分匹配数据示例: {eliminated_data[:2]}")
3. 注意事项与最佳实践
- GIL与多进程: 理解Python GIL是关键。对于CPU密集型任务，multiprocessing是比threading更好的选择，因为它绕过了GIL的限制，实现了真正的并行计算。
- 数据共享: 进程间数据共享比线程间复杂。普通Python对象在进程间默认不共享，需要使用multiprocessing.Manager创建共享数据结构，或者通过管道/队列进行通信。本例中Manager.list用于收集结果，避免了复杂的同步机制。
- 数据拷贝: 在eliminate_marking函数中，我们通过json_list[:]将json_list的浅拷贝传递给每个进程。这意味着每个进程都在自己的json_list副本上进行查找。这样做的好处是避免了对同一共享json_list进行并发读写和删除操作的复杂同步问题。缺点是，如果目标是修改原始的json_list（例如，从中删除匹配项），这种方法不会直接实现。但对于“获取数据”的需求，收集匹配结果是更安全和常见的模式。
- 任务分块: 合理地划分任务（marking_list的chunk_size）对性能至关重要。过小的块可能导致过多的进程创建和管理开销；过大的块可能导致部分进程负载不均。通常，进程数量不应超过CPU的核心数。可以根据实际情况调整chunk_size或直接根据os.cpu_count()来分配任务。
- 进程开销: 进程的创建和销毁比线程更耗资源。对于非常短小的任务，多进程的开销可能抵消并行带来的收益。但在本例这种大数据量、CPU密集型任务中，多进程的优势非常明显。
- 错误处理: 在生产环境中，应考虑在进程中加入错误处理机制，例如使用try-except块，并记录异常，以提高程序的健壮性。
- Manager的生命周期: 确保在所有子进程完成后调用manager.shutdown()来清理Manager创建的资源。
4. 总结
通过将任务分解为独立的子任务并在多个进程中并行执行，结合multiprocessing.Manager实现结果的有效收集，我们成功地将大数据量列表匹配和筛选的性能提升了一个数量级。这种方法特别适用于那些受限于Python GIL的CPU密集型计算任务。理解multiprocessing的工作原理和最佳实践，能够帮助开发者在处理大规模数据时构建出更高效、更健壮的Python应用程序。
