Python多处理模块

+ datasources
|-- __init__.py
|-- data_one.py
|-- data_two.py
|-- data_three.py
+ metrics
|-- __init__.py
|-- metric_one.py
|-- metric_two.py
+ outputs
|-- output_one.py
|-- output_two.py
- app.py

import datasources
import metrics
import outputs

for datasource in dir(datasources):
    datasource.refresh()
for metric in dir(metrics):
    metric.calculate()
for output in dir(outputs):
    output.dump()

data = []

def refresh():
    # Populate the "data" member somehow
    data = [1,2,3]
    return

import datasources.data_one as data_one
import datasources.data_two as data_two

data = []

def calculate():
    # Use the datasources to compute the metric
    data = [sum(x) for x in zip(data_one,data_two)]
    return

def run_thread(datasource):
    datasource.refresh()

threads = []
for datasource in dir(datasources):
    thread = threading.Thread(target=run_thread,args=(datasource))
    threads.append(thread)
    thread.start()
for thread in threads:
    thread.join()

def run_pool(calculate):
    calculate()

pool = multiprocessing.Pool()
pool.map(run_pool,[m.calculate for m in dir(metrics)]
pool.close()
pool.join()

metrics.metric_one.data

根据评论再次更新：
因为你在2.7,并且你正在处理模块而不是对象,所以你在挑选你需要的东西时遇到了问题.解决方法并不漂亮.它涉及将每个模块的名称传递给您的操作功能.我更新了部分部分,并且还更新了删除with语法.

一些东西：

首先,一般来说,多线程比线程更好.使用线程,您总是冒着处理Global Interpreter Lock的风险,这可能效率极低.如果您使用多核,这将成为一个非问题.

其次,你有正确的概念,但是通过拥有一个全局到模块的数据成员,你会让它变得奇怪.使您的源返回您感兴趣的数据,并使您的度量(和输出)将数据列表作为输入并输出结果列表.

这会将你的伪代码变成这样的东西：

app.py：

import datasources
import metrics
import outputs

pool = multiprocessing.Pool()
data_list = pool.map(lambda o: o.refresh,list(dir(datasources)))
pool.close()
pool.join()

pool = multiprocessing.Pool()
metrics_funcs = [(m,data_list) for m in dir(metrics)]
metrics_list = pool.map(lambda m: m[0].calculate(m[1]),metrics_funcs)
pool.close()
pool.join()

pool = multiprocessing.Pool()
output_funcs = [(o,data_list,metrics_list) for o in dir(outputs)]
output_list = pool.map(lambda o: o[0].dump(o[1],o[2]),output_funcs)
pool.close()
pool.join()

完成此操作后,您的数据源将如下所示：

def refresh():
    # Populate the "data" member somehow
    return [1,3]

您的指标看起来像这样：

def calculate(data_list):
    # Use the datasources to compute the metric
    return [sum(x) for x in zip(data_list)]

最后,您的输出可能如下所示：

def dump(data_list,metrics_list):
    # do whatever; you now have all the information

删除数据“全局”并传递它使每个部分更清洁(并且更容易测试).这突出了使每件作品完全独立.正如你所看到的,我正在做的就是改变传递给map的列表中的内容,在这种情况下,我通过将它们作为元组传递并在函数中解压缩来注入所有先前的计算.当然,你不必使用lambdas.您可以单独定义每个函数,但实际上并没有太多定义.但是,如果确实定义了每个函数,则可以使用partial函数来减少传递的参数数量.我经常使用这种模式,在你更复杂的代码中,你可能需要.这是一个例子：

from functools import partial

do_dump(module_name,metrics_list):
    globals()[module_name].dump(data_list,metrics_list)

invoke = partial(do_dump,data_list=data_list,metrics_list=metrics_list)
with multiprocessing.Pool() as pool:
    output_list = pool.map(invoke,[o.__name__ for o in dir(outputs)])
    pool.close()
    pool.join()

根据评论更新：

当您使用map时,您可以保证输入的顺序与输出的顺序相匹配,即data_list [i]是运行dir(datasources)[i] .refresh()的输出.我不会将数据源模块导入指标,而是将其更改为app.py：

data_list = ...
pool.close()
pool.join()
data_map = {name: data_list[i] for i,name in enumerate(dir(datasources))}

然后将data_map传递给每个指标.然后,度量标准按名称获取它想要的数据,例如

d1 = data_map['data_one']
d2 = data_map['data_two']
return [sum(x) for x in zip([d1,d2])]