用未来的运行关闭Google Cloud PubSub客户端

Question

def pubsub_callback(future):
    message_id = future.result()
    LOGGER.info("Successfully published %s", message_id)

def send_oneoff_pubsub_message(self, client=None):
    if not client:
        client = self.get_client('pubsubpub') # Creates a pubsub publisher client
    
    future = client.publish({...})
    try:
        future.exception(timeout=10)
    except Exception as exc:
        print("error")

    future.add_done_callback(pubsub_callback)

现在，在许多地方，我们正在缓慢地重构，以显式地在函数之外创建客户机（因此我们不会创建太多客户机）。然而，我仍然希望重构它，以便在消息发布后关闭客户端。

链接的问题建议在处理完客户端后client.api.transport._channel.close()。但是，在本例中，我只是在pubsub_callback被触发之后才完成它。

我看不出有任何方法可以从将来获得客户机，而且回调add_done_callback不允许发送参数。

Update:
查看代码，似乎这将在将来成功关闭客户端：

def send_oneoff_pubsub_message(self, client=None):
    if not client:
        client = self.get_client('pubsubpub') # Creates a pubsub publisher client

    future = client.publish({...})
    try:
        future.exception(timeout=10)
    except Exception as exc:
        print("error")

    future.add_done_callback(pubsub_callback)
    future.result(timeout=10)
    client.api.transport._channel.close()

这种asn方法有什么缺点吗？除了函数块直到发布（这对我来说是可以的）

Answer 1

您的代码混合执行异步操作和同步操作。在调用publish之后，您将等待长达10秒的时间来等待Future上的异常。大多数情况下，publish将在10秒内完成，因此在这一点上，您不妨同步调用future.result，甚至不必使用add_done_callback:

def pubsub_callback(future):
  try:
    message_id = future.result()
    LOGGER.info("Successfully published %s", message_id)
  except Exception as exc:
    print("error")

def send_oneoff_pubsub_message(self, client=None):
  if not client:
    client_created = True
    client = self.get_client('pubsubpub') # Creates a pubsub publisher client

  future = client.publish({...})
  pubsub_callback(future)

  if client_created:
    client.api.transport._channel.close()

如果希望异步执行，可以使用functools.partial:

from functools import partial

def pubsub_callback(client, future):
  try:
    message_id = future.result()
    LOGGER.info("Successfully published %s", message_id)
  except Exception as exc:
    print("error")

  if client:
    client.api.transport._channel.close()

def send_oneoff_pubsub_message(self, client=None):
  if not client:
    client_created = True
    client = self.get_client('pubsubpub') # Creates a pubsub publisher client

  if client_created:
    callback = partial(pubsub_callback, client)
  else:
    callback = partial(pubsub_callback, None)

  future = client.publish({...})
  future.add_done_callback(callback)

任何一种方式都应该允许您在所需的点关闭客户端。