如何使用 getattr 进行多线程，每个属性一个线程？

Question

假设我有以下具有多个昂贵属性的对象，如下所示：

class Object:

  def __init__(self, num):
    self.num = num

  @property 
  def expensive_property(self):
    return expensive_calculation

  @property 
  def expensive_property1(self):
    return expensive_calculation

  @property 
  def expensive_property2(self):
    return expensive_calculation

注意：昂贵房产的数量可能会随着时间的推移而增加。 给定Objects 的列表，我如何为列表中的所有对象计算每个线程的每个昂贵属性。我很难弄清楚我应该如何安排我的游泳池。

这就是我想要实现的目标：

from multithreading.dummy import Pool
from multithreading.dummy import Queue

object_list = [Object(i) for i in range(20)]
properties = [expensive_property2, expensive_propert5, expensive_property9, expensive_property3]


def get(obj, expensive_property):
  return [getattr(expensive_property, o) for o in obj]

tasks = Queue()
for p in properties :
  tasks.put((get, o, p))

results = []

with Pool(len(properties )) as pool:
   while True:
      task = tasks.get()
      if task is None:
        break
      func, *args = task
      result = pool.apply_async(func, args)
      results.append(result)

Answer 1

这有点疯狂，因为apply_async 有一个internal queue 在池中分配任务。我可以想象有理由围绕可观察性或背压设置另一个队列。你的例子是你的完整程序吗？还是您正在从不同的进程/线程排队工作？

如果您的计算受 CPU 限制，一个选项可能是删除队列以使事情变得更简单：

def wait_all(async_results, timeout_seconds_per_task=1):
   for r in async_results:
      r.get(timeout_seconds)

wait_all(
  [pool.apply_async(get, (o, p)) for p in properties],
  timeout_seconds_per_task=1,
)

就像上面的示例一样，这允许您在可用的 cpu 上分配计算（池甚至默认为 cpus on your machine 的数量）。如果您的工作受 IO 限制（由您的睡眠建议），则进程的收益可能会递减。

您必须进行基准测试，但对于 IO 绑定，您可以使用相同的模式创建线程池 https://stackoverflow.com/a/3034000/594589

其他选项可能是使用带有事件循环的非阻塞 IO，例如 gevent 或 asyncio。两者都允许您对相同的基于池的模式进行建模！