【问题标题】:How to make a Subscriber sleep if there are no messages from publisher in Google PUBSUB如果 Google PUBSUB 中没有来自发布者的消息,如何让订阅者休眠
【发布时间】:2020-03-26 18:14:24
【问题描述】:

当文档上传到 Google Cloud Storage Bucket 时,我正在使用 Google PubSub Publisher 触发通知并向订阅者发送消息。

我通过 -> gsutil notification create -t​​ [TOPIC_NAME] -f json -e OBJECT_FINALIZE gs://[BUCKET_NAME]

创建了一个通知主题

我的订阅者功能是: **

def callback(message):
    try:
        #storage_client = storage.Client.from_service_account_json('storage_service_key.json')
        print('Received message: {}'.format(message.data.decode("utf-8")))
        data = json.loads(message.data.decode("utf-8"))

        filename = data['name']
        file_name = re.search(r'/(.*)', filename).group(1)
        #filelink = data['selfLink']
        print("Processing the file : {}".format(file_name))
        path = "sample/"+file_name
        download_files(path, file_name)
        rming.image_remover(file_name) ## my custom function
        message.ack()
        os.remove(file_name)


    except Exception as error_message:
         print("Error in callback method: {}".format(error_message))
         pass


flow_control = pubsub_v1.types.FlowControl(max_messages=1)

subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(proj_name, sub_name)

streaming_pull_future = subscriber.subscribe(
    subscription_path, callback=callback,flow_control=flow_control
)
print('Listening for messages on: {}'.format(subscription_path))
with subscriber:
    try:
        # When `timeout` is not set, result() will block indefinitely,
        # unless an exception is encountered first.
        #flag=0
        streaming_pull_future.result()
    except Exception as error_message:  # noqa
        print("exception occured while handling subscription: {}").format(error_message)
        pass

** 我想将此订阅者功能部署在 docker 容器中。然后它将使订阅者 24/7 运行。

当没有可用消息时,我可以让订阅者休眠/(不发出拉取请求)到发布者吗?或者我的观点是,我可以让订阅者在没有传入消息的空闲时间休眠。

感谢任何帮助!!!!

【问题讨论】:

  • GCP Pub/Sub 有两种模式。推和拉。通过拉取,您基本上可以询问 GCP 是否有消息。如果没有可用的,您仍然会立即返回,下次拉动时由您决定。将其视为投票。第二个故事是订阅可以标记为推送。这意味着当有新消息可用时,GCP 通过 REST 调用调用您的应用程序并手动将消息传递给您。如果您的消息不频繁,这可能是您需要的。
  • 除了@Kolban 注释,不要实现拉取(轮询)。这会消耗 API 调用,您可以达到配额限制。
  • 感谢科尔班和约翰!!!我是 GCP 和 PUB/SUB 的新手,你们能给我一个可以在 python 中将订阅标记为推送的工作示例吗...
  • 另外一个查询将能够从 Publisher 提供的消息属性中获得与代码中使用的相同类型的属性,例如“name”、“'selfLink”,并且我能够使用带有回调和 flow_control 的订阅者?
  • 这里是推送文档的链接...cloud.google.com/pubsub/docs/push 如果您还在“gcp pubsub push”上进行谷歌搜索,您将获得大量点击,包括@987654322 上的一些精彩视频@

标签: docker google-cloud-platform publish-subscribe google-cloud-pubsub


【解决方案1】:

另外,总结一下cmets,还有很多东西

Cloud Run 缩放为 0,您可能只需要处理时间,四舍五入到 100 毫秒的上限。没有消息时它会休眠并且不花费任何费用。

【讨论】:

    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 2010-11-17
    • 2021-12-19
    • 1970-01-01
    • 2019-11-06
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    相关资源
    最近更新 更多