我很困惑做我想做的事情的最佳方式。
我想要什么?
- 对烧瓶应用程序的API调用
- Flask路由使用Process模块启动4-5个多进程,并使用共享的Managers().list()组合结果(在切片的pandas数据帧上)
- 将计算结果返回给客户端
我的实现:
pos_iter_list = get_chunking_iter_list(len(position_records), 10000)
manager = Manager()
data_dict = manager.list()
processes = []
for i in range(len(pos_iter_list) - 1):
temp_list = data_dict[pos_iter_list[i]:pos_iter_list[i + 1]]
p = Process(
target=transpose_dataset,
args=(temp_list, name_space, align_namespace, measure_master_id, df_searchable, products,
channels, all_cols, potential_col, adoption_col, final_segment, col_map, product_segments,
data_dict)
)
p.start()
processes.append(p)
for p in processes:
p.join()
我的目录结构:
- main.py(flask entry point)
- helper.py(contains function where above code is executed & calls transpose_dataset function)
运行相同程序时出错?RuntimeError:找不到所提供模块"的根路径mp_main";。这可能是因为模块来自一个不提供文件名信息的导入挂钩,或者是因为它是一个命名空间包。在这种情况下,需要显式地提供根路径。
不确定这里发生了什么,当使用if __name__ == '__main__':
从sample.py文件调用时,管理器列表工作正常
更新:同样的代码在我的MacBook上运行良好,而在windows操作系统上则不然。
API样品瓶调用:
@app.route(PREFIX + "ping", methods=['GET'])
def ping():
man = mp.Manager()
data = man.list()
processes = []
for i in range(0,5):
pr = mp.Process(target=test_func, args=(data, i))
pr.start()
processes.append(pr)
for pr in processes:
pr.join()
return json.dumps(list(data))
Stack有一个持续的错误阻止我发表评论,所以我只写一个答案。。
Python有两种(主要)方法来启动一个新进程:;"产卵";,以及";叉子";。Fork是一个仅在*nix(读作:linux或macos)中可用的系统命令,因此spawn是windows中唯一的选项。3.8之后的衍生将是MacOS上的默认版本,但fork仍然可用。最大的区别在于,fork基本上是对现有进程进行复制,而spawn则启动一个全新的进程(就像打开一个新的cmd窗口一样)。为什么以及如何运行有很多细微之处,但为了能够运行您希望子进程使用spawn运行的函数,子进程必须import
主文件。导入一个文件相当于只执行该文件,然后通常将其命名空间绑定到一个变量:import flask
将运行flak/__ini__.py文件,并将其全局命名空间绑定到变量flask
。然而,通常有一些代码只由主进程使用,不需要在子进程中导入/执行。在某些情况下,再次运行该代码实际上会中断工作,因此您需要防止它在主进程之外运行。考虑到这一点;魔术;变量__name__
仅等于主文件中的"__main__"
(在子进程或导入模块时不等于)。
在您的特定情况下,您将创建一个新的app = Flask(__name__)
,它在运行服务器之前进行一些验证和检查。当从子进程运行时,它会跳过这些设置/验证步骤之一。通过根本不让它运行来修复它是imao更干净的解决方案,但你也可以通过给它一个不会绊倒的值来修复它,然后永远不要启动辅助服务器(再次通过if __name__ == "__main__":
保护它)