python - 使用 python 的 Google Pubsub 模拟器

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