使用 Flask 在后台异步作业之上模拟同步请求

Question

我先解释一下我的系统的架构，然后转入问题：

我有一个 REST API，用作我的 API 网关。该服务器是使用 Flask 构建的。我也有 RabbitMQ 集群，以及我编写的一个客户端，它监听特定队列并执行其获取的任务。

到目前为止，我的所有请求都是异步的，所以一旦请求到达 API 网关，带有 URL 的 callback_uri 字段将作为请求的一部分提供结果，API 网关只负责用于将任务发送到 RabbitMQ，worker 处理任务，最后将结果 POST 回回调 URL。

我的问题是：

我希望新端点是同步的，处理仍将由与以前相同的工作人员完成，但我希望将结果返回 API 网关以返回给用户并释放连接。

我目前的解决方案：

我像以前一样将唯一的callback_uri 作为请求的一部分发送给工作人员，但现在特定端点由我的 API 网关实现，并允许 POST 和 GET 方法，因此工作人员可以发布一次结果它完成了，我的 API 网关不断轮询回调 URL，直到有结果可用，然后将结果返回给客户端。

除了让一个忙着等待的 HTTP 工作者轮询自己的端点以获取结果之外，还有其他首选选项吗？但仍然是同步的，所以只有在结果可用时才释放连接？

代码仅供说明：

@app.route('/long_task', methods=['POST'])
@sync_request
def long_task():
    try:
        if request.get_json() is None:
            return ERROR_MSG_NO_JSON, 400
        create_and_send_request_to_rabbitmq()
        return '', 200
    except Exception as ex:
        return ERROR_MSG_NO_DATA, 400


def sync_request(func):

    def call(*args, **kwargs):
        create_callback_uri()
        result = func(*args, **kwargs)
        status_code = result[1]
        if status_code == 200:
            result = get_callback_result()
        return result

    return call

def get_callback_result():
    callback_uri = request.get_json()['callback_uri']
    has_answer = False
    headers = {'content-type': 'application/json'}
    empty_response = {}
    content = json.dumps(empty_response)

    try:
        with Timeout(seconds=SYNC_REQUEST_TIMEOUT_SECONDS):
            while not has_answer:
                response = requests.get(callback_uri, headers=headers)
                if response.status_code == 200:
                    has_answer = True
                    content = response.content
                else:
                    time.sleep(0.2)
    except TimeoutException:
        log.debug('Timed out on sync request for request %s ' % request)

    return content, 200

Answer 1

因此，如果我理解正确，您希望您的后端等待某个工作人员的响应（通过 RabbitMQ）。您可以通过实现rpc over rabbitmq 来实现这一点。关键思想是使用相关 id。

但绝对最有效的方法是通过 websockets（或原始 tcp 套接字，如果它不是浏览器）运行客户端，并在工作完成时直接通知他。这样您就不会锁定资源（客户端连接、rabbitmq 队列）并避免性能损失 (rpc)。