如何在使用Flask的多进程时实现WSGI？

huangapple
go评论108阅读模式
英文:

How to implement WSGI while using multiprocessing in flask?

问题

以下是您要翻译的代码部分：

  1. 假设我有一个在函数中的视频处理程序，我想使用`multiprocessing`模块而不是`threading`来实现真正的并行处理。
  3. 因此，我的代码大致如下：
  4. ```python
  5. def start_subprocess(pid, progress_dict):
  6.     '''
  7.     假设这是视频处理的启动器...
  8.     '''
  9.     from time import sleep
  10.     # 模拟具有可变进度的子进程
  11.     progress = 0
  12.     while progress < 100:
  13.         sleep(1)
  14.         progress += 10
  15.         progress_dict[pid] = progress
  18. def get_current_progress_of_subprocess(pid, progress_dict):
  19.     '''
  20.     假设这是根据pid获取当前进度的视频当前进度，在这个上下文中，当前进度是所有已处理的当前帧...
  21.     '''
  22.     # 检索子进程的当前进度
  23.     if pid in progress_dict:
  24.         return progress_dict[pid]
  25.     else:
  26.         return None
  29. def flask_service(progress_dict):
  30.     from flask import Flask, request, jsonify
  31.     from multiprocessing import Process
  32.     app = Flask(__name__)
  34.     @app.route('/start_process')
  35.     def start_process():
  36.         pid = request.args.get('pid')
  37.         if pid is not None:
  38.             try:
  39.                 pid = int(pid)
  40.             except ValueError:
  41.                 return jsonify({'message': '无效的pid。'}), 400
  43.             # 启动新的子进程
  44.             if pid not in progress_dict:
  45.                 process = Process(target=start_subprocess, args=(pid, progress_dict))
  46.                 process.start()
  47.                 progress_dict[pid] = 0
  48.             else:
  49.                 return jsonify({'message': f'已启动pid {pid} 的进程。'}), 400
  51.             return jsonify({'message': f'已启动pid：{pid} 的进程。'}), 200
  52.         else:
  53.             return jsonify({'message': '未提供pid。'}), 400
  55.     @app.route('/get_progress')
  56.     def get_progress():
  57.         pid = request.args.get('pid')
  58.         if pid is not None:
  59.             try:
  60.                 pid = int(pid)
  61.             except ValueError:
  62.                 return jsonify({'message': '无效的pid。'}), 400
  64.             # 检索子进程的当前进度
  65.             current_progress = get_current_progress_of_subprocess(pid, progress_dict)
  67.             if current_progress is not None:
  68.                 return jsonify({'message': f'pid {pid} 的当前进度为{current_progress}。'}), 200
  69.             else:
  70.                 return jsonify({'message': f'未找到pid {pid} 的进程。'}), 404
  71.         else:
  72.             return jsonify({'message': '未提供pid。'}), 400
  74.     app.run(debug=False, threaded=True)
  77. if __name__ == '__main__':
  78.     from multiprocessing import Process, Manager
  79.     with Manager() as manager:
  80.         progress_dict = manager.dict()
  81.         p1 = Process(target=flask_service, args=(progress_dict,))
  82.         p1.start()
  83.         try:
  84.             p1.join()
  85.         except KeyboardInterrupt:
  86.             p1.terminate()
  87.             p1.join()
  88.         finally:
  89.             print('结束！')

我已经翻译了您的代码，以下是关于部署和并行处理的问题的回答：

您提到from flask import Flask创建了与WSGI兼容的实例。在部署时，您可以使用WSGI服务器（如Gunicorn、uWSGI等）来托管您的Flask应用程序。例如，使用Gunicorn，您可以运行以下命令来启动应用：

  1. gunicorn your_module:app

这将启动一个WSGI服务器，并使用app变量来处理Flask应用程序。

关于真正的并行处理，您使用了multiprocessing模块创建了多个进程，每个进程都可以并行处理视频处理任务。这是一种有效的并行处理方法，可以充分利用硬件资源，因此您确实实现了真正的并行处理，以解决视频处理等并行问题。

英文:

Suppose I have video process handler in a function, I want implement true parallel processing using multiprocessing module instead of threading.

So my code looks like this in general:

  1. def start_subprocess(pid, progress_dict):
  2. &#39;&#39;&#39;
  3. Suppose this is video processing starter...
  4. &#39;&#39;&#39;
  5. from time import sleep
  6. # Simulating a subprocess with variable progress
  7. progress = 0
  8. while progress &lt; 100:
  9. sleep(1)
  10. progress += 10
  11. progress_dict[pid] = progress
  12. def get_current_progress_of_subprocess(pid, progress_dict):
  13. &#39;&#39;&#39;
  14. Suppose this is video current progress by pid, in this context current progress are all current frames has been processed... 
  15. &#39;&#39;&#39;
  16. # Retrieve current progress of a subprocess
  17. if pid in progress_dict:
  18. return progress_dict[pid]
  19. else:
  20. return None
  21. def flask_service(progress_dict):
  22. from flask import Flask, request, jsonify
  23. from multiprocessing import Process
  24. app = Flask(__name__)
  25. @app.route(&#39;/start_process&#39;)
  26. def start_process():
  27. pid = request.args.get(&#39;pid&#39;)
  28. if pid is not None:
  29. try:
  30. pid = int(pid)
  31. except ValueError:
  32. return jsonify({&#39;message&#39;: f&#39;Invalid pid.&#39;}), 400
  33. # Start a new subprocess
  34. if pid not in progress_dict:
  35. process = Process(target=start_subprocess, args=(pid, progress_dict))
  36. process.start()
  37. progress_dict[pid] = 0
  38. else:
  39. return jsonify({&#39;message&#39;: f&#39;Process with pid {pid} already started.&#39;}), 400
  40. return jsonify({&#39;message&#39;: f&#39;Process started with pid: {pid}&#39;}), 200
  41. else:
  42. return jsonify({&#39;message&#39;: &#39;No pid provided.&#39;}), 400
  43. @app.route(&#39;/get_progress&#39;)
  44. def get_progress():
  45. pid = request.args.get(&#39;pid&#39;)
  46. if pid is not None:
  47. try:
  48. pid = int(pid)
  49. except ValueError:
  50. return jsonify({&#39;message&#39;: f&#39;Invalid pid.&#39;}), 400
  51. # Retrieve current progress of the subprocess
  52. current_progress = get_current_progress_of_subprocess(pid, progress_dict)
  53. if current_progress is not None:
  54. return jsonify({&#39;message&#39;: f&#39;Current progress of pid: {pid} is {current_progress}.&#39;}), 200
  55. else:
  56. return jsonify({&#39;message&#39;: f&#39;Process with pid {pid} not found.&#39;}), 404
  57. else:
  58. return jsonify({&#39;message&#39;: &#39;No pid provided.&#39;}), 400
  59. app.run(debug=False, threaded=True)
  60. if __name__ == &#39;__main__&#39;:
  61. from multiprocessing import Process, Manager
  62. with Manager() as manager:
  63. progress_dict = manager.dict()
  64. p1 = Process(target=flask_service, args=(progress_dict,))
  65. p1.start()
  66. try:
  67. p1.join()
  68. except KeyboardInterrupt:
  69. p1.terminate()
  70. p1.join()
  71. finally:
  72. print(&#39;Ending up!&#39;)

I have achieved what I want, but the problem is how do I deploy this with WSGI? As far As I know from flask import Flask class is creating instance of compatible WSGI. So what it looks like in deployment?

Also, am I actually implement true parallel processing? I just want make sure if I really do it. true parallel I mean is using hardware capabilities to solve parallel issue such as video processing.

答案1

得分: 1

  1. def create_app():
  2.   from flask import Flask, request, jsonify
  3.   from multiprocessing import Process, Manager
  5.   app = Flask(__name__)
  7.   # 创建受管理的字典：
  8.   progress_dict = Manager().dict()
  10.   @app.route('/start_process')
  11.   def start_process():
  12.     pid = request.args.get('pid')
  13.     if pid is not None:
  14.       try:
  15.         pid = int(pid)
  16.       except ValueError:
  17.         return jsonify({'message': f'无效的pid。'}), 400
  19.       # 启动新的子进程
  20.       if pid not in progress_dict:
  21.         process = Process(target=start_subprocess, args=(pid, progress_dict))
  22.         process.start()
  23.         progress_dict[pid] = 0
  24.       else:
  25.         return jsonify({'message': f'已经启动了pid为{pid}的进程。'}), 400
  27.       return jsonify({'message': f'已启动进程，pid为：{pid}。'}), 200
  28.     else:
  29.       return jsonify({'message': '未提供pid。'}), 400
  31.   @app.route('/get_progress')
  32.   def get_progress():
  33.     pid = request.args.get('pid')
  34.     if pid is not None:
  35.       try:
  36.         pid = int(pid)
  37.       except ValueError:
  38.         return jsonify({'message': f'无效的pid。'}), 400
  40.       # 检索子进程的当前进度
  41.       current_progress = get_current_progress_of_subprocess(pid, progress_dict)
  43.       if current_progress is not None:
  44.         return jsonify({'message': f'pid为{pid}的当前进度为{current_progress}。'}), 200
  45.       else:
  46.         return jsonify({'message': f'未找到pid为{pid}的进程。'}), 404
  47.     else:
  48.       return jsonify({'message': '未提供pid。'}), 400
  50.   return app
英文:

I think your code is more complicated than it needs to be. You should create a function, create_app, which creates the Flask application and the process_dict managed dictionary and then returns the Flask application. Functionget_current_progress_of_subprocess remains unchanged and I added handling of a KeyboardInterrupt in start_subprocess.

  1. def start_subprocess(pid, progress_dict):
  2.   &#39;&#39;&#39;
  3.   Suppose this is video processing starter...
  4.   &#39;&#39;&#39;
  5.   from time import sleep
  6.   # Simulating a subprocess with variable progress
  7.   try:
  8.     progress = 0
  9.     while progress &lt; 100:
  10.       sleep(1)
  11.       progress += 10
  12.       progress_dict[pid] = progress
  13.   except KeyboardInterrupt:
  14.     pass
  17. def get_current_progress_of_subprocess(pid, progress_dict):
  18.   &#39;&#39;&#39;
  19.   Suppose this is video current progress by pid, in this context current progress are all current frames has been processed...
  20.   &#39;&#39;&#39;
  21.   # Retrieve current progress of a subprocess
  22.   if pid in progress_dict:
  23.     return progress_dict[pid]
  24.   else:
  25.     return None
  28. def create_app():
  29.   from flask import Flask, request, jsonify
  30.   from multiprocessing import Process, Manager
  32.   app = Flask(__name__)
  34.   # Create the managed dictionary:
  35.   progress_dict = Manager().dict()
  37.   @app.route(&#39;/start_process&#39;)
  38.   def start_process():
  39.     pid = request.args.get(&#39;pid&#39;)
  40.     if pid is not None:
  41.       try:
  42.         pid = int(pid)
  43.       except ValueError:
  44.         return jsonify({&#39;message&#39;: f&#39;Invalid pid.&#39;}), 400
  46.       # Start a new subprocess
  47.       if pid not in progress_dict:
  48.         process = Process(target=start_subprocess, args=(pid, progress_dict))
  49.         process.start()
  50.         progress_dict[pid] = 0
  51.       else:
  52.         return jsonify({&#39;message&#39;: f&#39;Process with pid {pid} already started.&#39;}), 400
  54.       return jsonify({&#39;message&#39;: f&#39;Process started with pid: {pid}&#39;}), 200
  55.     else:
  56.       return jsonify({&#39;message&#39;: &#39;No pid provided.&#39;}), 400
  58.   @app.route(&#39;/get_progress&#39;)
  59.   def get_progress():
  60.     pid = request.args.get(&#39;pid&#39;)
  61.     if pid is not None:
  62.       try:
  63.         pid = int(pid)
  64.       except ValueError:
  65.         return jsonify({&#39;message&#39;: f&#39;Invalid pid.&#39;}), 400
  67.       # Retrieve current progress of the subprocess
  68.       current_progress = get_current_progress_of_subprocess(pid, progress_dict)
  70.       if current_progress is not None:
  71.         return jsonify({&#39;message&#39;: f&#39;Current progress of pid: {pid} is {current_progress}.&#39;}), 200
  72.       else:
  73.         return jsonify({&#39;message&#39;: f&#39;Process with pid {pid} not found.&#39;}), 404
  74.     else:
  75.       return jsonify({&#39;message&#39;: &#39;No pid provided.&#39;}), 400
  77.   return app
  80. if __name__ == &#39;__main__&#39;:
  81.   create_app().run(debug=False, threaded=True)

huangapple
  • 本文由 发表于 2023年5月21日 03:13:13
  • 转载请务必保留本文链接：https://go.coder-hub.com/76296957.html
  • deployment
  • flask
  • multiprocessing
  • python
  • wsgi
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定