编辑:更新了环境信息(请参见第一部分)
环境
我正在使用Python 2.7
Ubuntu 16.04
问题
我有一个应用程序,我已将其简化为三个阶段的过程:
- 从多个数据源(HTTP请求、系统信息等)收集数据
- 基于这些数据计算指标
- 以各种格式输出这些指标
每个阶段必须在进入下一个阶段之前完成,但是每个阶段由可以并行运行的多个子任务组成(我可以发送3个HTTP请求并读取系统日志,同时等待它们返回)
我已将这些阶段划分为模块,并将子任务划分为子模块,因此我的项目层次结构如下:
+ 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
`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()
“额外的代码包装了dir调用以忽略系统模块,进行异常处理等等——但这就是它的要点。”
“每个数据源子模块大致如下:”
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()
这个可以工作,之后我可以计算任何指标,
datasources.x.data
属性会被填充。为了并行化第二阶段(指标),因为它更少依赖于I/O而更多依赖于CPU,我感觉简单的线程并不能真正加速,我需要使用multiprocessing模块来利用多个核心。我写了以下代码:
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
它返回
[]
,就像从未运行该模块一样。不知何故,使用多进程模块似乎会将线程作用域限制,以至于它们不再共享数据属性。我应该如何重写这个代码,以便可以并行计算每个指标,利用多个核心,但在完成后仍然可以访问数据?