模拟在与瓶

Question

后台异步工作顶部同步请求我先解释一下我的系统的体系结构，然后移动到这样一个问题：

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

到目前为止，我的所有请求都是异步的，所以一旦请求到达API网关，一个带有URL的callback_uri字段将POST结果作为请求的一部分提供，API网关只负责将任务发送到RabbitMQ，工作人员处理完任务，并在结束时将结果回送到回调URL。

我的问题是：

我希望有一个新的端点是在感同步，该处理将通过同一个工作像以前仍然完成，但我想要得到的结果反馈给API网关返回给用户并释放连接。

我目前的解决方案：

我发送一个唯一的callback_uri作为请求对工人的一部分，但现在具体的终结点我的API网关实现，同时允许POST和GET方法，所以工作人员可以在完成后POST结果，并且我的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）。