您可以使用 Workflows 连接器 来支持 Pub/Sub 操作,包括将消息发布到 Pub/Sub 主题
Pub/Sub 主题是向其发送消息的资源 发布商。订阅代表要传送到订阅应用的主题的消息流。详细了解 Pub/Sub。
发布消息
创建 Pub/Sub 主题并订阅该主题后,您可以创建一个工作流,将消息发布到该主题:
YAML
- init: assign: - project: '${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}' - topic: TOPIC_ID - subscription: SUBSCRIPTION_ID - message: hello: world - base64Msg: '${base64.encode(json.encode(message))}' - publish_message_to_topic: call: googleapis.pubsub.v1.projects.topics.publish args: topic: '${"projects/" + project + "/topics/" + topic}' body: messages: - data: '${base64Msg}'
JSON
[ { "init": { "assign": [ { "project": "${sys.get_env(\"GOOGLE_CLOUD_PROJECT_ID\")}" }, { "topic": "TOPIC_ID" }, { "subscription": "SUBSCRIPTION_ID" }, { "message": { "hello": "world" } }, { "base64Msg": "${base64.encode(json.encode(message))}" } ] } }, { "publish_message_to_topic": { "call": "googleapis.pubsub.v1.projects.topics.publish", "args": { "topic": "${\"projects/\" + project + \"/topics/\" + topic}", "body": { "messages": [ { "data": "${base64Msg}" } ] } } } } ]
替换以下内容:
TOPIC_ID
:Pub/Sub 主题的 ID 或完全限定标识符。SUBSCRIPTION_ID
:ID 或完全限定的标识符 Pub/Sub 订阅。
拉取消息
您可以创建 Eventarc 触发器,以将 Pub/Sub 主题连接到 Workflows 事件接收器。将消息发布到 Pub/Sub 主题来生成事件,该事件将作为 为目标工作流添加运行时参数。如需了解详情,请参阅使用事件或 Pub/Sub 消息触发工作流。
您还可以创建用于拉取 Pub/Sub 消息的工作流。在以下示例中,工作流会等待消息发布 轮询。
YAML
- pullMessage: call: googleapis.pubsub.v1.projects.subscriptions.pull args: subscription: '${"projects/" + project + "/subscriptions/" + subscription}' body: maxMessages: 1 result: m - checkState: switch: - condition: ${m.receivedMessages[0].message.data != ""} next: outputMessage - wait: call: sys.sleep args: seconds: 60 next: pullMessage - outputMessage: return: '${json.decode(base64.decode(m.receivedMessages[0].message.data))}'
JSON
[ { "pullMessage": { "call": "googleapis.pubsub.v1.projects.subscriptions.pull", "args": { "subscription": "${\"projects/\" + project + \"/subscriptions/\" + subscription}", "body": { "maxMessages": 1 } }, "result": "m" } }, { "checkState": { "switch": [ { "condition": "${m.receivedMessages[0].message.data != \"\"}", "next": "outputMessage" } ] } }, { "wait": { "call": "sys.sleep", "args": { "seconds": 60 }, "next": "pullMessage" } }, { "outputMessage": { "return": "${json.decode(base64.decode(m.receivedMessages[0].message.data))}" } } ]