在处理数据时，往往会涉及到一个数据需要进行多次加工，这时候我们一般是通过Pipeline的方式进行处理。那么在Knative Eventing中是否也能支持对一个事件进行分步骤多次处理？这个还真有。从 0.7 版本开始，Knative Eventing中提供了一个 Sequence 资源模型，可用于事件Pipeline处理。

Sequence 定义

首先我们看一下Sequence Spec定义：

apiVersion: messaging.knative.dev/v1alpha1
kind: Sequence
metadata:
 name: test
spec:
 channelTemplate:
 apiVersion: messaging.knative.dev/v1alpha1
 kind: InMemoryChannel
 steps:
 - ref:
 apiVersion: serving.knative.dev/v1alpha1
 kind: Service
 name: test
 reply:
 kind: Broker
 apiVersion: eventing.knative.dev/v1alpha1
 name: test

Sequence Spec包括3个部分：

steps：在step中定义了按照顺序执行的服务，每个服务会对应创建Subscription。
channelTemplate：指定了使用具体的那个Channel
reply：（可选）定义了最后一个step返回结果的响应目标

在 Broker/Trigger 模型中使用 Sequence

我们将创建以下逻辑配置。创建一个 cronjobsource，向 Broker 提供事件，然后创建一个 filter，将这些事件连接到由3个 step 组成的 Sequence 中。然后，我们获取最后的step返回结果事件发送给给Broker，并创建另一个 Trigger，该 Trigger 随后将显示事件结果。

对于这个例子，这里设置一个 Broker 程序、一个 InMemoryChannel 以及一个 Knative Service（用于显示事件结果）。示例使用 default namespace。

如果要使用不同类型的Channel，则需要修改
sequence.spec.channeltemplate以创建对应的 Channel 资源。

创建 Knative Service

首先创建3个Knative Service，用于 Sequence 中服务处理

apiVersion: serving.knative.dev/v1alpha1
kind: Service
metadata:
 name: first
spec:
 template:
 spec:
 containers:
 - image: us.gcr.io/probable-summer-223122/cmd-03315b715ae8f3e08e3a9378df706fbb@sha256:2656f39a7fcb6afd9fc79e7a4e215d14d651dc674f38020d1d18c6f04b220700
 env:
 - name: STEP
 value: "0"
---
apiVersion: serving.knative.dev/v1alpha1
kind: Service
metadata:
 name: second
spec:
 template:
 spec:
 containers:
 - image: us.gcr.io/probable-summer-223122/cmd-03315b715ae8f3e08e3a9378df706fbb@sha256:2656f39a7fcb6afd9fc79e7a4e215d14d651dc674f38020d1d18c6f04b220700
 env:
 - name: STEP
 value: "1"
---
apiVersion: serving.knative.dev/v1alpha1
kind: Service
metadata:
 name: third
spec:
 template:
 spec:
 containers:
 - image: us.gcr.io/probable-summer-223122/cmd-03315b715ae8f3e08e3a9378df706fbb@sha256:2656f39a7fcb6afd9fc79e7a4e215d14d651dc674f38020d1d18c6f04b220700
 env:
 - name: STEP
 value: "2"
---

执行创建命令：

kubectl -n default create -f ./steps.yaml

创建 Sequence

创建Sequence，这里依次顺序执行[first->second->third]这3个服务。将最终处理的结果发送到broker-test中。

apiVersion: messaging.knative.dev/v1alpha1
kind: Sequence
metadata:
 name: sequence
spec:
 channelTemplate:
 apiVersion: messaging.knative.dev/v1alpha1
 kind: InMemoryChannel
 steps:
 - ref:
 apiVersion: serving.knative.dev/v1alpha1
 kind: Service
 name: first
 - ref:
 apiVersion: serving.knative.dev/v1alpha1
 kind: Service
 name: second
 - ref:
 apiVersion: serving.knative.dev/v1alpha1
 kind: Service
 name: third
 reply:
 kind: Broker
 apiVersion: eventing.knative.dev/v1alpha1
 name: broker-test

执行如下命令：

kubectl -n default create -f ./sequence.yaml

创建CronJobSource指向Broker

这里将创建一个 cronjobsource，它将每2分钟发送一个{"message": "Hello world!"} 信息到 broker-test 中。

apiVersion: sources.eventing.knative.dev/v1alpha1
kind: CronJobSource
metadata:
 name: cronjob-source
spec:
 schedule: "*/2 * * * *"
 data: '{"message": "Hello world!"}'
 sink:
 apiVersion: eventing.knative.dev/v1alpha1
 kind: Broker
 name: broker-test

执行命令如下：

kubectl -n default create -f ./cron-source.yaml

为Sequence创建Trigger

创建订阅事件类型为：dev.knative.cronjob.event 的 Trigger, 用于Sequence 进行消费处理。

apiVersion: eventing.knative.dev/v1alpha1
kind: Trigger
metadata:
 name: sequence-trigger
spec:
 filter:
 sourceAndType:
 type: dev.knative.cronjob.event
 subscriber:
 ref:
 apiVersion: messaging.knative.dev/v1alpha1
 kind: Sequence
 name: sequence

执行如下命令：

kubectl -n default create -f ./trigger.yaml

创建结果订阅 Trigger

创建结果订阅 Trigger，订阅samples.http.mod3 的事件类型，对 sequence 执行的结果进行显示

apiVersion: serving.knative.dev/v1alpha1
kind: Service
metadata:
 name: sequence-display
spec:
 template:
 spec:
 containers:
 - image: gcr.io/knative-releases/github.com/knative/eventing-sources/cmd/event_display
---
apiVersion: eventing.knative.dev/v1alpha1
kind: Trigger
metadata:
 name: sequence-trigger
spec:
 filter:
 sourceAndType:
 type: samples.http.mod3
 subscriber:
 ref:
 apiVersion: serving.knative.dev/v1alpha1
 kind: Service
 name: sequence-display
---