文档教程云原生【免费下载链接】docsUser documentation for Knative components.项目地址https://gitcode.com/gh_mirrors/docs1/docs点击查看免费下载导读本文基于 Knative 官方文档仓库中的 sequence-with-broker-trigger 示例讲解如何将 Knative Eventing 的三种核心能力——Broker事件总线、Trigger订阅过滤、Sequence有序处理流水线——组合成一个完整的端到端事件处理链路。读完本文你将掌握如何让PingSource定时事件进入 Broker如何用 Trigger 将事件按type路由进一个三步 Sequence以及如何让 Sequence 的处理结果重新回到 Broker 供其他 Trigger 消费并展示。文中所有 YAML 均可在仓库对应目录中找到可直接运行的版本。一、逻辑架构事件如何流经 Broker 与 Sequence本示例要构建的逻辑配置如下图所示PingSource产生事件并送入Broker一个Triggersequence-trigger按事件type过滤后把事件交给由 3 个步骤组成的Sequence依次处理Sequence 末尾reply将新生成了内容的事件重新送回Broker最后另一个Triggerdisplay-trigger订阅这批处理结果并投递给sequence-display服务打印到日志。PingSource ──► Broker(default) ──► Trigger(sequence-trigger) ──► Sequence(3 步) │ ▼ reply sequence-display ◄── Trigger(display-trigger) ◄── Broker(default)从源码结构看这是一个闭环式的事件流设计Sequence 的产出不是终点而是作为新事件重新进入事件总线供更多消费者订阅。这正是 Broker/Trigger 模型相对点对点直连的价值——解耦了事件的生产、处理与消费Sequence只关心处理逻辑本身。示例中用到的转换函数是 Knative Eventing 自带的appender命令位于 eventing 仓库的cmd/appender它对每个经过的事件在data.message后追加一段固定文本用来直观验证事件确实按顺序经过了每一个步骤。二、前置条件在开始之前需要确认集群环境满足以下要求已安装Knative Serving用于部署作为 Sequence 步骤与事件展示端的 Knative Service已安装Knative Eventing并具备可用的InMemoryChannelSequence 底层依赖 Channel 在步骤间传递事件示例默认在default命名空间中执行所有命令中的default均可替换为实际命名空间。!!! note 关于 Channel 类型 若你希望使用其他类型的Channel例如 Kafka Channel需要修改Sequence.Spec.ChannelTemplate让 Sequence 创建对应的 Channel 资源。InMemoryChannel适合演示与开发环境生产环境应评估更可靠的 Channel 实现。!!! info 关于 hairpin 流量 Knative 官方文档在 Sequence 总览 中特别提醒Sequence 依赖hairpin回环流量即 Pod 需要能够通过自身 Service IP 访问自己。若集群不支持该特性通常属于 CNI 层面的集群级配置需要联系集群管理员处理。三、分步部署完整流程下面按照官方文档的顺序从零开始部署整个链路。每一步都对应仓库中的一份 YAML 文件。3.1 创建 BrokerdefaultBroker是 Knative Eventing 的事件总线所有事件先汇聚于此。Knative Eventing 默认提供MTChannelBasedBroker其默认底层 Channel 即InMemoryChannel参见 Creating a Broker。创建一个名为default的集群默认 Broker将以下 YAML 保存为文件例如broker.yamlapiVersion: eventing.knative.dev/v1 kind: Broker metadata: name: default应用该文件kubectl apply -f filename.yaml其中filename为上一步创建的文件名。创建完成后可校验状态kubectl -n default get broker default当 Broker 正常就绪时READY状态为True并会得到一个形如http://broker-ingress.knative-eventing.svc.cluster.local/default/default的地址。3.2 创建三个 Knative Service 作为 Sequence 步骤Sequence 的每个 step 都需要是一个Addressable可寻址对象。这里使用三个 KnativeService全部基于appender镜像通过MESSAGE环境变量配置各自要追加的文本Service 名称MESSAGE 值额外配置作用first - Handled by 0—第 1 步追加文本 0second - Handled by 1—第 2 步追加文本 1third - Handled by 2TYPE: samples.http.mod3第 3 步追加文本 2并改写事件type其中第三步通过TYPE环境变量将处理后事件的 CloudEventtype改写为samples.http.mod3——这是后续display-trigger过滤该事件的关键标记。将以下 YAML 保存为steps.yamlapiVersion: serving.knative.dev/v1 kind: Service metadata: name: first spec: template: spec: containers: - image: gcr.io/knative-releases/knative.dev/eventing/cmd/appender env: - name: MESSAGE value: - Handled by 0 --- apiVersion: serving.knative.dev/v1 kind: Service metadata: name: second spec: template: spec: containers: - image: gcr.io/knative-releases/knative.dev/eventing/cmd/appender env: - name: MESSAGE value: - Handled by 1 --- apiVersion: serving.knative.dev/v1 kind: Service metadata: name: third spec: template: spec: containers: - image: gcr.io/knative-releases/knative.dev/eventing/cmd/appender env: - name: MESSAGE value: - Handled by 2 - name: TYPE value: samples.http.mod3 ---将default替换为你配置 Broker 的命名空间然后创建kubectl -n default create -f ./steps.yaml3.3 创建 Sequence串联三个步骤并回写 BrokerSequence是 flows.knative.dev/v1 下的 CRD用于定义一组按顺序依次执行的函数。官方文档在 Sequence 总览 中明确了其 Spec 的三个部分steps按顺序执行的Subscriber列表每个 step 须为 Addressable 对象语法与Subscription的SubscriberSpec一致channelTemplateSequence 内部在各步骤之间创建Channel时使用的模板reply可选Sequence 最后一步的输出结果发送到的目标引用。Sequence 会在底层自动创建若干Channel和Subscription来串联各步骤。其Status也相应包含四部分Conditions整体状态、ChannelStatuses各步骤间 Channel 的状态数组第 n 项对应第 n 步之前的 Channel、SubscriptionStatuses连接各 Channel 与步骤的 Subscription 状态数组、以及AddressStatusSequence 对外暴露的地址发送到该地址即进入第一步之前的 Channel。本示例的关键在于spec.reply指向了defaultBroker——Sequence 第三步的输出会作为新事件回写进 Broker。将以下 YAML 保存为sequence.yamlapiVersion: flows.knative.dev/v1 kind: Sequence metadata: name: sequence spec: channelTemplate: apiVersion: messaging.knative.dev/v1 kind: InMemoryChannel steps: - ref: apiVersion: serving.knative.dev/v1 kind: Service name: first - ref: apiVersion: serving.knative.dev/v1 kind: Service name: second - ref: apiVersion: serving.knative.dev/v1 kind: Service name: third reply: ref: kind: Broker apiVersion: eventing.knative.dev/v1 name: default说明若使用其他 Channel 类型请修改spec.channelTemplateapiVersion与kind需对应你选择的 Channel CRD例如 Kafka Channelspec.reply.name必须指向你实际创建的 Broker 名称。创建 Sequencekubectl -n default create -f ./sequence.yaml3.4 创建 PingSource定时产生事件并送入 BrokerPingSource是 Knative Eventing 内置的事件源按 cron 调度产生固定负载的事件参考 Creating a PingSource object 与 PingSource reference。本示例中它每 2 分钟发送一个 data 为{message: Hello world!}的 CloudEventsink指向defaultBroker。将以下 YAML 保存为ping-source.yamlapiVersion: sources.knative.dev/v1 kind: PingSource metadata: name: ping-source spec: schedule: */2 * * * * contentType: application/json data: {message: Hello world!} sink: ref: apiVersion: eventing.knative.dev/v1 kind: Broker name: default关键字段说明详见 PingSource referencespec.schedulecron 调度表达式默认* * * * *每分钟支持标准 cron 与带秒字段的 Quartz 格式spec.contentTypedata的媒体类型如application/json默认空spec.data/spec.dataBase64事件体二者互斥二进制数据需用dataBase64传 base64 编码字符串spec.sink事件目标可为任意 Addressable 对象的引用。创建 PingSourcekubectl -n default create -f ./ping-source.yamlPingSource 产生的事件type为dev.knative.sources.ping这正是下一个 Trigger 的过滤依据。3.5 创建 Trigger将 PingSource 事件路由进 SequenceTrigger表示对某个 Broker 事件的订阅意愿参考 Using Triggers。本示例的sequence-trigger订阅defaultBroker通过filter.attributes精确匹配事件type dev.knative.sources.ping将命中事件投递给名为sequence的 Sequence。将以下 YAML 保存为trigger.yamlapiVersion: eventing.knative.dev/v1 kind: Trigger metadata: name: sequence-trigger spec: broker: default filter: attributes: type: dev.knative.sources.ping subscriber: ref: apiVersion: flows.knative.dev/v1 kind: Sequence name: sequence注意这里的subscriber.ref.kind是Sequence而非Service——Trigger 的订阅目标可以是任何实现了Addressable契约的对象Sequence 通过其AddressStatus暴露地址指向第一步之前的 Channel因此可以被直接订阅。filter.attributes属于官方文档中描述的Legacy attributes filter对任意数量的 CloudEvent 属性与扩展做精确匹配语义等同exact方言大小写敏感。对于 MTChannelBasedBroker还支持filters字段下的exact、prefix、suffix、all、any、not、cesql等多种过滤方言当filters与filter同时存在时filters优先。创建 Triggerkubectl -n default create -f ./trigger.yaml3.6 创建展示服务与第二个 Trigger消费 Sequence 的处理结果为了让 Sequence 的处理结果可见创建sequence-display服务基于event_display镜像将收到的 CloudEvent 完整打印到日志并创建display-trigger订阅 Broker 中type samples.http.mod3的事件——这正是第三步third服务通过TYPE环境变量改写后的类型因此只有经过了完整 Sequence 的事件才会被投递给展示服务。将以下 YAML 保存为display-trigger.yamlapiVersion: serving.knative.dev/v1 kind: Service metadata: name: sequence-display spec: template: spec: containers: - image: gcr.io/knative-releases/knative.dev/eventing/cmd/event_display --- apiVersion: eventing.knative.dev/v1 kind: Trigger metadata: name: display-trigger spec: broker: default filter: attributes: type: samples.http.mod3 subscriber: ref: apiVersion: serving.knative.dev/v1 kind: Service name: sequence-display ---创建服务与 Triggerkubectl -n default create -f ./display-trigger.yaml至此完整链路已部署完毕。所有 YAML 均可在仓库目录 sequence-with-broker-trigger 中找到包括 steps.yaml、sequence.yaml、ping-source.yaml、trigger.yaml 和 display-trigger.yaml。四、验证结果从日志中读懂事件流转4.1 查看 Podkubectl -n default get pods4.2 查看 sequence-display 的日志kubectl -n default logs -l serving.knative.dev/servicesequence-display -c user-container --tail-1等待一到两个调度周期后输出应类似于☁️ cloudevents.Event Validation: valid Context Attributes, specversion: 1.0 type: samples.http.mod3 source: /apis/v1/namespaces/default/pingsources/ping-source id: 159bba01-054a-4ae7-b7be-d4e7c5f773d2 time: 2020-03-03T14:56:00.000652027Z datacontenttype: application/json Extensions, knativearrivaltime: 2020-03-03T14:56:00.018390608Z knativehistory: default-kne-trigger-kn-channel.default.svc.cluster.local; sequence-kn-sequence-0-kn-channel.default.svc.cluster.local; sequence-kn-sequence-1-kn-channel.default.svc.cluster.local; sequence-kn-sequence-2-kn-channel.default.svc.cluster.local; default-kne-trigger-kn-channel.default.svc.cluster.local traceparent: 00-e893412106ff417a90a5695e53ffd9cc-5829ae45a14ed462-00 Data, { id: 0, message: Hello world! - Handled by 0 - Handled by 1 - Handled by 2 }4.3 日志解读这条日志直观地印证了整条链路的正确性type: samples.http.mod3说明事件确实被第三步third服务改写了类型从而通过了display-trigger的过滤Data.message初始的Hello world!被依次追加了- Handled by 0、- Handled by 1、- Handled by 2证明事件按顺序经过了first→second→third三个步骤knativehistory扩展分号分隔的列表完整记录了事件途径的 Channel 链——先是 Broker 的default-kne-trigger-kn-channel对应 sequence-trigger 的投递接着是 Sequence 内部各步骤间的sequence-kn-sequence-0/1/2-kn-channel最后又回到 Broker 的default-kne-trigger-kn-channel对应 display-trigger 的投递。这条历史正是事件在 Broker 与 Sequence 间往返流动的底层证据source保持为 PingSource 的地址说明经过 Sequence 处理后事件保留了来源信息。五、原理深入Sequence 在底层做了什么结合官方文档与上文示例可以总结 Sequence 的运作机制自动创建 Channel 与 Subscription每两个相邻步骤之间Sequence 都会按spec.channelTemplate的模板创建一个Channel并创建一个Subscription把前一步的输出接到下一步。三步 Sequence 内部因此会形成Channel0 → first → Channel1 → second → Channel2 → third的链条Reply 决定终点spec.reply指定最后一步输出送往何处。本示例将其指向 Broker实现处理结果回流事件总线在 sequence-terminal 示例 中不设 reply末步为终态在 sequence-reply-to-event-display 示例 中则直接送往展示服务在 sequence-reply-to-sequence 示例 中甚至可以链到另一个 SequenceAddressable 契约Sequence 通过AddressStatus暴露地址使得它既可以作为 Trigger 的subscriber也可以作为其他事件源/Subscription 的sink——这正是本示例中 Trigger 能直接订阅 Sequence 的原因。六、清理资源验证结束后可按需清理# 删除本示例创建的资源按需执行 kubectl -n default delete sequence sequence kubectl -n default delete trigger sequence-trigger display-trigger kubectl -n default delete pingsource ping-source kubectl -n default delete service.serving.knative.dev first second third sequence-display kubectl -n default delete broker default更简单的做法是删除整个命名空间若资源都部署在其中kubectl delete namespace namespace七、进一步探索本文展示的是 Sequence 与 Broker/Trigger 组合的标准范式。若想深入了解相邻主题仓库中还有以下资料可供参考Sequence 总览与完整示例列表涵盖 Spec/Status 详解以及无 reply终态、reply 到展示服务、Sequence 链式串联等变体Eventing Flows 总览Sequence有序函数列表与 Parallel分支并行分发两类流式 CRD 的对比Creating a BrokerBroker 类型、默认配置与创建方式Using TriggersTrigger 过滤方言与注解eventing.knative.dev/injection、knative.dev/dependencyCreating a PingSource object 与 PingSource reference事件源的完整字段说明。掌握事件 → Broker → Trigger → Sequence → 回写 Broker → 再消费这一闭环后你便可以在 Knative 上组合出更复杂的异步事件处理架构——例如把 Sequence 的每个步骤替换为真实的业务服务、把回写目标换成其他 Broker 或下游系统从而在保持各环节解耦的同时获得有序、可观测的事件处理流水线。赞分享文档教程云原生【免费下载链接】docsUser documentation for Knative components.项目地址https://gitcode.com/gh_mirrors/docs1/docs点击查看免费下载相关推荐Knative Sequence 事件流编排实战从顺序管道到 Broker/Trigger 集成Knative Sequence 事件流编排实战从顺序管道到 Broker/Trigger 集成 Sequence 是 Knative Eventing 提供文档教程云原生使用 Knative Eventing Flows 编排事件流Sequence 与 Parallel 实战指南使用 Knative Eventing Flows 编排事件流Sequence 与 Parallel 实战指南 Knative Eventing 提供了一组文档教程云原生Knative Eventing 事件驱动架构实战指南Source、Broker、Trigger 与事件流Knative Eventing 事件驱动架构实战指南Source、Broker、Trigger 与事件流 Knative Eventing 是 Knativ文档教程云原生上一篇mathjs 数据序列化指南利用 JSON、replacer 与 reviver 持久化与传输数学对象下一篇可比较公司分析Comps技能实战指南从可比公司筛选到机构级 Excel 估值工作簿创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
