【发布时间】：2021-01-28 01:23:42
【问题描述】：
TLDR： 我需要为多处理设置一个烧瓶应用程序，以便 API 和 stomp 队列侦听器在单独的进程中运行，因此不会干扰彼此的操作。
详情： 我正在构建一个具有 API 端点的 python 烧瓶应用程序，并且还创建一个消息队列侦听器以使用 stomp 包连接到一个 activemq 队列。
我需要实现多处理，以便 API 和侦听器不会阻止彼此的操作。这样，API 将接受新请求，并且侦听器将继续侦听新消息并相应地执行任务。
代码的简化版本如下所示（为简洁起见，省略了一些细节）。
问题：多处理导致应用程序卡住。 worker 的 run 方法没有被一致地调用，因此监听器永远不会被创建。
# Start the worker as a subprocess -- this is not working -- app gets stuck before the worker's run method is called
m = Manager()
shared_state = m.dict()
worker = MyWorker(shared_state=shared_state)
worker.start()
经过几天的故障排除后，我怀疑问题是由于未正确设置多处理。我能够证明是这样，因为当我剥离所有多处理代码并直接调用工作者的 run 方法时，所有队列管理代码都正常工作，CustomWorker 模块创建侦听器，创建消息，并拿起消息。我认为这表明队列管理代码工作正常，问题的根源很可能是多处理造成的。
# Removing the multiprocessing and calling the worker's run method directly works without getting stuck so the issue is likely due to multiprocessing not being setup correctly
worker = MyWorker()
worker.run()
这是我目前的代码：
应用程序
这部分代码创建 API 并尝试创建一个新进程来创建队列侦听器。 'custom_worker_utils' 模块是一个自定义模块，它在 CustomWorker() 类的 run 方法中创建 stomp 监听器。
from flask import Flask, request, make_response, jsonify
from flask_restx import Resource, Api
import sys, os, logging, time
basedir = os.path.dirname(os.getcwd())
sys.path.append('..')
from custom_worker_utils.custom_worker_utils import *
from multiprocessing import Manager
# app.py
def create_app():
app = Flask(__name__)
app.config['BASE_DIR'] = basedir
api = Api(app, version='1.0', title='MPS Worker', description='MPS Common Worker')
logger = get_logger()
'''
This is a placeholder to trigger the sending of a message to the first queue
'''
@api.route('/initialapicall', endpoint="initialapicall", methods=['GET', 'POST', 'PUT', 'DELETE'])
class InitialApiCall(Resource):
#Sends a message to the queue
def get(self, *args, **kwargs):
mqconn = get_mq_connection()
message = create_queue_message(initial_tracker_file)
mqconn.send('/queue/test1', message, headers = {"persistent":"true"})
return make_response(jsonify({'message': 'Initial Test Call Worked!'}), 200)
# Start the worker as a subprocess -- this is not working -- app gets stuck before the worker's run method is called
m = Manager()
shared_state = m.dict()
worker = MyWorker(shared_state=shared_state)
worker.start()
# Removing the multiprocessing and calling the worker's run method directly works without getting stuck so the issue is likely due to multiprocessing not being setup correctly
#worker = MyWorker()
#worker.run()
return app
自定义工作者工具
run() 方法被调用，连接到队列并使用 stomp 包创建监听器
# custom_worker_utils.py
from multiprocessing import Manager, Process
from _datetime import datetime
import os, time, json, stomp, requests, logging, random
'''
The listener
'''
class MyListener(stomp.ConnectionListener):
def __init__(self, p):
self.process = p
self.logger = p.logger
self.conn = p.mqconn
self.conn.connect(_user, _password, wait=True)
self.subscribe_to_queue()
def on_message(self, headers, message):
message_data = json.loads(message)
ticket_id = message_data[constants.TICKET_ID]
prev_status = message_data[constants.PREVIOUS_STEP_STATUS]
task_name = message_data[constants.TASK_NAME]
#Run the service
if prev_status == "success":
resp = self.process.do_task(ticket_id, task_name)
elif hasattr(self, 'revert_task'):
resp = self.process.revert_task(ticket_id, task_name)
else:
resp = True
if (resp):
self.logger.debug('Acknowledging')
self.logger.debug(resp)
self.conn.ack(headers['message-id'], self.process.conn_id)
else:
self.conn.nack(headers['message-id'], self.process.conn_id)
def on_disconnected(self):
self.conn.connect('admin', 'admin', wait=True)
self.subscribe_to_queue()
def subscribe_to_queue(self):
queue = os.getenv('QUEUE_NAME')
self.conn.subscribe(destination=queue, id=self.process.conn_id, ack='client-individual')
def get_mq_connection():
conn = stomp.Connection([(_host, _port)], heartbeats=(4000, 4000))
conn.connect(_user, _password, wait=True)
return conn
class CustomWorker(Process):
def __init__(self, **kwargs):
super(CustomWorker, self).__init__()
self.logger = logging.getLogger("Worker Log")
log_level = os.getenv('LOG_LEVEL', 'WARN')
self.logger.setLevel(log_level)
self.mqconn = get_mq_connection()
self.conn_id = random.randrange(1,100)
for k, v in kwargs.items():
setattr(self, k, v)
def revert_task(self, ticket_id, task_name):
# If the subclass does not implement this,
# then there is nothing to undo so just return True
return True
def run(self):
lst = MyListener(self)
self.mqconn.set_listener('queue_listener', lst)
while True:
pass
【问题讨论】：
-
“我需要实现多处理，以便 API 和侦听器不会阻塞彼此的操作。” => 创建一个 Flask 程序来处理 API 请求。为您的听众创建一个完全独立的程序。这两个程序（或进程）应该通过套接字/管道进行通信，因为这看起来就像您设置的那样。基本上不要试图劫持烧瓶进程并让它产生额外的进程，就像你发现的那样，它不是为那样工作而设计的。
-
@steve 非常感谢！您能否提供更多有关进程应如何通过套接字/管道进行通信的信息？
-
进程通过进程间通信 (IPC) 进行通信 查看 wiki/google 以获取有关不同类型 IPC 的更多信息。两种类型的 IPC 是套接字/管道。 ActiveMQ 是一种通过管道/套接字证明 IPC 的机制。基本上你想完全分离你的 Flask 代码和 stomp 监听器。它们应该是完全独立的程序，然后通过 IPC 进行通信。因此，您可以从命令行运行您的烧瓶应用程序。然后你会单独运行你的监听器。
-
构建两个通过 API 通信的微服务
标签： python flask multiprocessing message-queue stomp