pubsub 路由实现了基于内容的路由,这是一种使用 DSL 而不是命令式应用程序代码的消息模式。通过 pubsub 路由,您可以根据 CloudEvents 的内容,将消息路由到应用程序中的不同 URI/路径和事件处理程序。如果没有匹配的路由,则可以使用可选的默认路由。随着您的应用程序扩展以支持多个事件版本或特殊情况,这种方法将非常有用。
虽然可以通过代码实现路由,但将路由规则与应用程序分离可以提高可移植性。
此功能适用于声明式和编程式订阅方法,但不适用于流式订阅。
对于声明式订阅,使用 dapr.io/v2alpha1
作为 apiVersion
。以下是使用路由的 subscriptions.yaml
示例:
apiVersion: dapr.io/v2alpha1
kind: Subscription
metadata:
name: myevent-subscription
spec:
pubsubname: pubsub
topic: inventory
routes:
rules:
- match: event.type == "widget"
path: /widgets
- match: event.type == "gadget"
path: /gadgets
default: /products
scopes:
- app1
- app2
在编程方法中,返回的是 routes
结构而不是 route
。JSON 结构与声明式 YAML 匹配:
import flask
from flask import request, jsonify
from flask_cors import CORS
import json
import sys
app = flask.Flask(__name__)
CORS(app)
@app.route('/dapr/subscribe', methods=['GET'])
def subscribe():
subscriptions = [
{
'pubsubname': 'pubsub',
'topic': 'inventory',
'routes': {
'rules': [
{
'match': 'event.type == "widget"',
'path': '/widgets'
},
{
'match': 'event.type == "gadget"',
'path': '/gadgets'
},
],
'default': '/products'
}
}]
return jsonify(subscriptions)
@app.route('/products', methods=['POST'])
def ds_subscriber():
print(request.json, flush=True)
return json.dumps({'success':True}), 200, {'ContentType':'application/json'}
app.run()
const express = require('express')
const bodyParser = require('body-parser')
const app = express()
app.use(bodyParser.json({ type: 'application/*+json' }));
const port = 3000
app.get('/dapr/subscribe', (req, res) => {
res.json([
{
pubsubname: "pubsub",
topic: "inventory",
routes: {
rules: [
{
match: 'event.type == "widget"',
path: '/widgets'
},
{
match: 'event.type == "gadget"',
path: '/gadgets'
},
],
default: '/products'
}
}
]);
})
app.post('/products', (req, res) => {
console.log(req.body);
res.sendStatus(200);
});
app.listen(port, () => console.log(`consumer app listening on port ${port}!`))
[Topic("pubsub", "inventory", "event.type ==\"widget\"", 1)]
[HttpPost("widgets")]
public async Task<ActionResult<Stock>> HandleWidget(Widget widget, [FromServices] DaprClient daprClient)
{
// Logic
return stock;
}
[Topic("pubsub", "inventory", "event.type ==\"gadget\"", 2)]
[HttpPost("gadgets")]
public async Task<ActionResult<Stock>> HandleGadget(Gadget gadget, [FromServices] DaprClient daprClient)
{
// Logic
return stock;
}
[Topic("pubsub", "inventory")]
[HttpPost("products")]
public async Task<ActionResult<Stock>> HandleProduct(Product product, [FromServices] DaprClient daprClient)
{
// Logic
return stock;
}
package main
import (
"encoding/json"
"fmt"
"log"
"net/http"
"github.com/gorilla/mux"
)
const appPort = 3000
type subscription struct {
PubsubName string `json:"pubsubname"`
Topic string `json:"topic"`
Metadata map[string]string `json:"metadata,omitempty"`
Routes routes `json:"routes"`
}
type routes struct {
Rules []rule `json:"rules,omitempty"`
Default string `json:"default,omitempty"`
}
type rule struct {
Match string `json:"match"`
Path string `json:"path"`
}
// This handles /dapr/subscribe
func configureSubscribeHandler(w http.ResponseWriter, _ *http.Request) {
t := []subscription{
{
PubsubName: "pubsub",
Topic: "inventory",
Routes: routes{
Rules: []rule{
{
Match: `event.type == "widget"`,
Path: "/widgets",
},
{
Match: `event.type == "gadget"`,
Path: "/gadgets",
},
},
Default: "/products",
},
},
}
w.WriteHeader(http.StatusOK)
json.NewEncoder(w).Encode(t)
}
func main() {
router := mux.NewRouter().StrictSlash(true)
router.HandleFunc("/dapr/subscribe", configureSubscribeHandler).Methods("GET")
log.Fatal(http.ListenAndServe(fmt.Sprintf(":%d", appPort), router))
}
<?php
require_once __DIR__.'/vendor/autoload.php';
$app = \Dapr\App::create(configure: fn(\DI\ContainerBuilder $builder) => $builder->addDefinitions(['dapr.subscriptions' => [
new \Dapr\PubSub\Subscription(pubsubname: 'pubsub', topic: 'inventory', routes: (
rules: => [
('match': 'event.type == "widget"', path: '/widgets'),
('match': 'event.type == "gadget"', path: '/gadgets'),
]
default: '/products')),
]]));
$app->post('/products', function(
#[\Dapr\Attributes\FromBody]
\Dapr\PubSub\CloudEvent $cloudEvent,
\Psr\Log\LoggerInterface $logger
) {
$logger->alert('Received event: {event}', ['event' => $cloudEvent]);
return ['status' => 'SUCCESS'];
}
);
$app->start();
在这些示例中,根据 event.type
,应用程序将被调用于:
/widgets
/gadgets
/products
表达式是用通用表达式语言 (CEL)编写的,其中 event
代表云事件。表达式中可以引用 CloudEvents 核心规范中的任何属性。
匹配“重要”消息:
has(event.data.important) && event.data.important == true
匹配大于 $10,000 的存款:
event.type == "deposit" && int(event.data.amount) > 10000
匹配消息的多个版本:
event.type == "mymessage.v1"
event.type == "mymessage.v2"
作为参考,以下属性来自 CloudEvents 规范。
根据术语 data 的定义,CloudEvents 可能 包含有关事件发生的领域特定信息。当存在时,此信息将被封装在 data
中。
datacontenttype
属性指定(例如 application/json),并在这些相应属性存在时遵循 dataschema
格式。以下属性在所有 CloudEvents 中是必需的:
String
source
+ id
对于每个不同的事件都是唯一的。如果由于网络错误而重新发送重复事件,则它可能具有相同的 id
。消费者可以假设具有相同 source
和 id
的事件是重复的。类型: URI-reference
描述: 标识事件发生的上下文。通常包括以下信息:
URI 中编码的数据的确切语法和语义由事件生产者定义。
生产者 必须 确保 source
+ id
对于每个不同的事件都是唯一的。
应用程序可以:
source
,以便更容易生成唯一的 ID,并防止其他生产者具有相同的 source
。source
标识符。一个 source 可能包含多个生产者。在这种情况下,生产者 必须 合作以确保 source
+ id
对于每个不同的事件都是唯一的。
约束:
示例:
类型: String
描述: 事件使用的 CloudEvents 规范版本。这使得上下文的解释成为可能。合规的事件生产者 必须 在引用此版本的规范时使用 1.0
值。
目前,此属性仅包含“主要”和“次要”版本号。这允许在不更改此属性值的情况下对规范进行补丁更改。
注意:对于“候选发布”版本,可能会使用后缀进行测试。
约束:
String
type
的版本信息。有关更多信息,请参阅CloudEvents 的版本控制。以下属性在 CloudEvents 中是可选的。有关可选定义的更多信息,请参阅符号约定部分。
类型: String
根据 RFC 2046
描述: data
值的内容类型。此属性使 data
能够携带任何类型的内容,其中格式和编码可能与所选事件格式不同。
例如,使用 JSON 信封格式呈现的事件可能在 data
中携带 XML 负载。此属性被设置为 "application/xml"
,通知消费者。
不同 datacontenttype
值的数据内容呈现规则在事件格式规范中定义。例如,JSON 事件格式在第 3.1 节中定义了关系。
对于某些二进制模式协议绑定,此字段直接映射到相应协议的内容类型元数据属性。您可以在相应协议中找到二进制模式和内容类型元数据映射的规范规则。
在某些事件格式中,您可以省略 datacontenttype
属性。例如,如果 JSON 格式事件没有 datacontenttype
属性,则意味着 data
是符合 "application/json"
媒体类型的 JSON 值。换句话说:没有 datacontenttype
的 JSON 格式事件与 datacontenttype="application/json"
的事件完全等效。
当将没有 datacontenttype
属性的事件消息转换为不同格式或协议绑定时,目标 datacontenttype
应明确设置为源的隐含 datacontenttype
。
约束:
有关媒体类型示例,请参阅 IANA 媒体类型
URI
data
遵循的模式。与模式不兼容的更改应通过不同的 URI 反映。有关更多信息,请参阅CloudEvents 的版本控制。类型: String
描述: 这描述了事件生产者(由 source
标识)上下文中的事件主题。在发布-订阅场景中,订阅者通常会订阅由 source
发出的事件。如果 source
上下文具有内部子结构,则仅 source
标识符可能不足以作为任何特定事件的限定符。
在上下文元数据中(而不是仅在 data
负载中)识别事件的主题在通用订阅过滤场景中很有帮助,其中中间件无法解释 data
内容。在上述示例中,订阅者可能只对名称以 ‘.jpg’ 或 ‘.jpeg’ 结尾的 blob 感兴趣。使用 subject
属性,您可以为该事件子集构建简单而高效的字符串后缀过滤器。
约束:
示例:
订阅者可能会注册对在 blob 存储容器中创建新 blob 时的兴趣。在这种情况下:
source
标识订阅范围(存储容器)type
标识“blob 创建”事件id
唯一标识事件实例,以区分同名 blob 的单独创建事件。新创建的 blob 的名称在 subject
中传递:
source
: https://example.com/storage/tenant/containersubject
: mynewfile.jpgTimestamp
source
的生产者 必须 在这方面保持一致。换句话说,要么他们都使用事件发生的实际时间,要么他们都使用相同的算法来确定使用的值。观看此视频以了解如何使用 pubsub 进行消息路由: