Google Pubsub模拟器与python

Question

有没有人有一个非常基本的使用模拟器的 Python 发布/订阅示例。

这是我的订户代码

## setup subscribers
from google.cloud import pubsub
print("subscribing to topic")

subscriber = pubsub.SubscriberClient()
subscription_path = subscriber.subscription_path(app.config['PUB_SUB_PROJECT'], app.config['PUB_SUB_TOPIC'])
def callback(message):
print('Received message: {}'.format(message))

subscriber.subscribe(subscription_path, callback=callback)

然后这是我的发布代码

from google.cloud import pubsub
publisher = pubsub.PublisherClient()
topic_path = publisher.topic_path(app.config['PUB_SUB_PROJECT'], app.config['PUB_SUB_TOPIC'])
try:
topic = publisher.create_topic(topic_path)
except Exception:
app.logger.info("Topic already exists")
data = "ein test"
data = data.encode('utf-8')
publisher.publish(topic_path, data=data)
print("published topic")

似乎发布有效 ->但我认为它实际上是发布到云队列而不是模拟器。因此，我的订户从未收到任何东西。

欢迎任何提示和技巧。我相信这就像确保发布者发布到模拟器并且订阅者从模拟器读取一样简单。

Answer 1

在 Python 中，您无需进行任何代码更改即可使用模拟器。相反，您必须定义PUBSUB_EMULATOR_HOST和PUBSUB_PROJECT_ID环境变量。

设置它们的最简单方法是在启动程序之前运行$(gcloud beta emulators pubsub env-init)。 如果您在本地使用 Google App Engine，请运行该命令，然后使用dev_appserver.py app.yaml --env_var PUBSUB_EMULATOR_HOST=${PUBSUB_EMULATOR_HOST}启动您的应用。

这记录在 https://cloud.google.com/pubsub/docs/emulator