在 grpc python 中处理异步流请求

我试图了解如何使用双向流处理 grpc api（使用 Python API）。

假设我有以下简单的服务器定义：

syntax = "proto3";
package simple;

service TestService {
  rpc Translate(stream Msg) returns (stream Msg){}
}

message Msg
{
 string msg = 1;
}

假设从客户端发送的消息是异步发送的（由于用户选择了一些 ui 元素）。

为客户端生成的 python 存根将包含一个方法Translate它将接受一个生成器函数并返回一个迭代器。

我不清楚的是如何编写生成器函数来返回用户创建的消息。在等待消息时在线程上休眠听起来并不是最好的解决方案。

现在这有点笨拙，但您可以按如下方式完成您的用例：

#!/usr/bin/env python

from __future__ import print_function

import time
import random
import collections
import threading

from concurrent import futures
from concurrent.futures import ThreadPoolExecutor
import grpc

from translate_pb2 import Msg
from translate_pb2_grpc import TestServiceStub
from translate_pb2_grpc import TestServiceServicer
from translate_pb2_grpc import add_TestServiceServicer_to_server


def translate_next(msg):
    return ''.join(reversed(msg))


class Translator(TestServiceServicer):
  def Translate(self, request_iterator, context):
    for req in request_iterator:
      print("Translating message: {}".format(req.msg))
      yield Msg(msg=translate_next(req.msg))

class TranslatorClient(object):
  def __init__(self):
    self._stop_event = threading.Event()
    self._request_condition = threading.Condition()
    self._response_condition = threading.Condition()
    self._requests = collections.deque()
    self._last_request = None
    self._expected_responses = collections.deque()
    self._responses = {}

  def _next(self):
    with self._request_condition:
      while not self._requests and not self._stop_event.is_set():
        self._request_condition.wait()
      if len(self._requests) > 0:
        return self._requests.popleft()
      else:
        raise StopIteration()

  def next(self):
    return self._next()

  def __next__(self):
    return self._next()

  def add_response(self, response):
    with self._response_condition:
      request = self._expected_responses.popleft()
      self._responses[request] = response
      self._response_condition.notify_all()

  def add_request(self, request):
    with self._request_condition:
      self._requests.append(request)
      with self._response_condition:
        self._expected_responses.append(request.msg)
      self._request_condition.notify()

  def close(self):
    self._stop_event.set()
    with self._request_condition:
      self._request_condition.notify()

  def translate(self, to_translate):
    self.add_request(to_translate)
    with self._response_condition:
      while True:
        self._response_condition.wait()
        if to_translate.msg in self._responses:
          return self._responses[to_translate.msg]


def _run_client(address, translator_client):
  with grpc.insecure_channel('localhost:50054') as channel:
    stub = TestServiceStub(channel)
    responses = stub.Translate(translator_client)
    for resp in responses:
      translator_client.add_response(resp)

def main():
  server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
  add_TestServiceServicer_to_server(Translator(), server)
  server.add_insecure_port('[::]:50054')
  server.start()
  translator_client = TranslatorClient()
  client_thread = threading.Thread(
      target=_run_client, args=('localhost:50054', translator_client))
  client_thread.start()

  def _translate(to_translate):
    return translator_client.translate(Msg(msg=to_translate)).msg

  translator_pool = futures.ThreadPoolExecutor(max_workers=4)
  to_translate = ("hello", "goodbye", "I", "don't", "know", "why",)
  translations = translator_pool.map(_translate, to_translate)
  print("Translations: {}".format(zip(to_translate, translations)))

  translator_client.close()
  client_thread.join()
  server.stop(None)


if __name__ == "__main__":
  main()

基本思想是有一个名为TranslatorClient在单独的线程上运行，关联请求和响应。它期望响应将按照请求发送的顺序返回。它还实现了迭代器接口，以便您可以将其直接传递给Translate存根上的方法。

我们启动一个正在运行的线程_run_client从中提取响应TranslatorClient并在另一端反馈它们add_response.

The main我在这里包含的函数实际上只是一个稻草人，因为我没有您的 UI 代码的详细信息。我在跑_translate in a ThreadPoolExecutor为了证明这一点，尽管translator_client.translate是同步的，它会产生，允许您同时处理多个正在进行的请求。

我们认识到，对于这样一个简单的用例，需要编写大量代码。最终的答案将是asyncio支持。我们在不久的将来对此有计划。但就目前而言，无论您运行的是 python 2 还是 python 3，这种解决方案都应该能让您继续下去。