如何用Pool以Pythonic且高效的方式并行过滤基于属性的对象列表？

场景与现有实现 #

我们需要过滤TestClass对象列表中满足self.x > self.y的对象，因数据量庞大、过滤逻辑资源消耗高，计划通过multiprocessing.Pool实现并行处理。现有实现代码如下：

import random
from multiprocessing import Pool

class TestClass:
    def __init__(self, x, y):
        self.x = x
        self.y = y

def filter(test_object):
    if test_object.x > test_object.y:
        return test_object
    else:
        return None

def parallel_process(operation, input, pool):
    result = pool.map(operation, input)
    return result

if __name__ == "__main__":
    N = 10
    list_of_objects = []
    for i in range(N):
        x = random.randint(0, 10)
        y = random.randint(0, 10)
        tst = TestClass(x, y)
        list_of_objects.append(tst)

    process_count = 2
    process_pool = Pool(process_count)
    result = parallel_process(filter, list_of_objects, process_pool)
    result = [i for i in result if i is not None]

现有方案的问题与优化空间 #

现有方案能实现需求，但并非兼具Pythonic风格与高效性的最优方案，主要存在以下可优化点：

• 避免覆盖内置函数
  自定义函数命名为filter会覆盖Python内置的filter()函数，属于不良编码习惯，建议改为语义化名称，比如should_keep或filter_x_gt_y。

• 简化对象创建逻辑
  原代码用循环逐个append生成对象列表，可替换为列表推导式，更简洁且符合Pythonic风格：

  list_of_objects = [TestClass(random.randint(0, 10), random.randint(0, 10)) for _ in range(N)]
  

• 优化进程池管理
  原代码手动创建进程池但未显式关闭，建议使用上下文管理器with自动管理进程池生命周期，避免资源泄漏：

  with Pool(process_count) as process_pool:
      # 并行操作逻辑
      pass
  

• 提升并行处理效率
  原方案返回整个对象或None，进程间传递对象需要序列化（pickle），若对象体积较大，会带来额外性能开销。更高效的方式是让过滤函数返回布尔值，仅传递少量数据，再通过布尔值与原列表配对筛选结果：

  def should_keep(obj):
      return obj.x > obj.y
  
  with Pool(process_count) as process_pool:
      keep_flags = process_pool.map(should_keep, list_of_objects)
      result = [obj for obj, keep in zip(list_of_objects, keep_flags) if keep]
  

  这种方式减少了进程间传输的数据量，序列化开销更低，整体效率更高。

• 冗余函数简化
  原代码中的parallel_process函数仅简单封装pool.map，无额外逻辑，可直接在主逻辑中调用pool.map，减少不必要的函数嵌套。

优化后的完整代码 #

import random
from multiprocessing import Pool

class TestClass:
    def __init__(self, x, y):
        self.x = x
        self.y = y

def should_keep(obj):
    return obj.x > obj.y

if __name__ == "__main__":
    N = 10
    list_of_objects = [TestClass(random.randint(0, 10), random.randint(0, 10)) for _ in range(N)]

    process_count = 2
    with Pool(process_count) as process_pool:
        keep_flags = process_pool.map(should_keep, list_of_objects)
        result = [obj for obj, keep in zip(list_of_objects, keep_flags) if keep]

总结 #

现有方案可行，但优化后的版本更符合Pythonic风格（简洁的列表推导式、语义化命名、上下文管理器），同时通过减少进程间数据传输提升了效率，是更优的实现方案。

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