在 Google Cloud Python SDK 中使用 AsyncClients 时，future 附加到不同的循环

Question

我对在 Python 中使用

asyncio

还很陌生。原来我有同步功能

speak

:

from google.cloud import texttospeech
client = texttospeech.TextToSpeechClient()
def speak(*args):
  # omitted
  result = client.synthesize_speech(...)

async def my_work(*args):
  # omitted
  loop = asyncio.get_running_loop()
  responses = ["Hi", "Bye"] # actually awaited results
  await asyncio.gather(*(
    loop.run_in_executor(
      None,
      speak,
      ...
    ) for text in responses
  ))

我没有遇到该设置的错误。 但现在我尝试使用异步版本：

from google.cloud import texttospeech
client = texttospeech.TextToSpeechAsyncClient()
async def speak(*args):
  # omitted
  result = await client.synthesize_speech(...)

async def my_work(*args):
  # omitted
  loop = asyncio.get_running_loop()
  responses = ["Hi", "Bye"] # actually awaited results
  await asyncio.gather(*[speak(...) for text in responses])

发生以下错误：

RuntimeError: Task
<Task pending name='Task-58' coro=<speak() running at /app/google_tts.py:36>
cb=[gather.<locals>._done_callback() at /usr/local/lib/python3.11/asyncio/tasks.py:764]>
got Future <Task pending name='Task-69' coro=<UnaryUnaryCall._invoke() running at /usr/local/lib/python3.11/site-packages/grpc/aio/_call.py:568>>
attached to a different loop

一些问题：

• 除了代码清晰之外，我想做的事情还有其他好处吗？

• running sync function elsewhere

  和

  running async counterparts

  我该如何选择？

Answer 1

根据https://cloud.google.com/python/docs/reference/texttospeech/latest/multiprocessing的信息，在 multiprocessing.pool.Pool 调用 os.fork() 后创建客户端。

from google.cloud import texttospeech
client = texttospeech.TextToSpeechAsyncClient()
async def speak(client, *args):
  # omitted
  result = await client.synthesize_speech(...)

async def my_work(*args):
  # omitted
  loop = asyncio.get_running_loop()
  responses = ["Hi", "Bye"] # actually awaited results
  with ThreadPoolExecutor() as pool:
    client = texttospeech.TextToSpeechAsyncClient()
    await asyncio.gather(*[speak(client, ...) for text in responses])