14-部署Kafkasource和KafkaChannel-Toy模板网

这篇具有很好参考价值的文章主要介绍了14-部署Kafkasource和KafkaChannel。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

部署KafkaSource

KafkaSource负责将Kafka中的消息记录转为CloudEvents
仅在需要从Kafka中加载消息并提供给Knative Eventing上的应用程序使用时才需要KafkaSource

命令：

kubectl apply -f https://github.com/knative-extensions/eventing-kafka-broker/releases/download/knative-v1.12.1/eventing-kafka-controller.yaml

kubectl apply -f https://mirror.ghproxy.com/https://github.com/knative-extensions/eventing-kafka-broker/releases/download/knative-v1.12.1/eventing-kafka-source.yaml

部署KafkaChannel

负责在Knative Eventing上提供基于Kafka集群的Channel实现，后端基于Kafka Topic

https://knative.dev/docs/install/yaml-install/eventing/install-eventing-with-yaml/#verify-the-installation

kubectl apply -f https://github.com/knative-extensions/eventing-kafka-broker/releases/download/knative-v1.12.1/eventing-kafka-controller.yaml

kubectl apply -f https://github.com/knative-extensions/eventing-kafka-broker/releases/download/knative-v1.12.1/eventing-kafka-channel.yaml

如果所使用的Kafka集群是外置的，那需要修改kafka-channel-config的configmap中boostrap server的地址.
```
kubectl get cm kafka-channel-config -n knative-eventing -o yaml
```

修改Default Channel的default-ch-webhook相关配置: 可以在全局修改，也可以在某个名称空间修改

apiVersion: v1
kind: ConfigMap
metadata:
  name: default-ch-webhook
  namespace: knative-eventing
data:
  default-ch-config: |
    clusterDefault:
      apiVersion: messaging.knative.dev/v1
      kind: InMemoryChannel
    namespaceDefaults:
      event-demo:
        apiVersion: messaging.knative.dev/v1beta1
        kind: KafkaChannel
        spec:
          numPartitions: 5
          replicationFactor: 1 # Kafka集群的节点数量，决定了其复制因子的可用值上限；但使用的partition数量可按需定义；

测试KafkaChannel

创建KafkaChannel

~$ kn channel create kc01 --type messaging.knative.dev:v1beta1:KafkaChannel

~$ kn channel list 
NAME    TYPE              URL                                                 AGE     READY   REASON 
kc01    KafkaChannel      http://kc01-kn-channel.default.svc.cluster.local    37s     True

提示：在没有设置KafkaChannel为默认Channel类型的名称空间下创建KafkaChannel时需要显式指定其类型。

测试KafkaChannel

创建订阅至KafkaChannel/kc01的Subscription

kn subscription create sub-to-kc01 --channel messaging.knative.dev:v1beta1:KafkaChannel:kc01 --sink ksvc:event-display

说明：此处事先存在一个可用的名为event-display的Knative Service。

创建用于模拟Source的Pod

kubectl run client-$RANDOM --image=ikubernetes/admin-box:v1.2 --rm --restart=Never -it --command -- /bin/bash

在Pod的交互式接口发送CloudEvents

root@client-18448 ~# curl -v "http://kc01-kn-channel.default.svc.cluster.local" -X POST -H "Content-Type: application/cloudevents+json" \
    -d '{"id": "100001", "specversion": "1.0", "type": "com.icloud2native.flow.seq", "source": "Curl", "data": {"message":"Hello from Knative Eventing KafkaChannel"}}'

在最后的Sink上查看生成的CloudEvents文章来源地址https://www.toymoban.com/news/detail-804957.html

POD=$(kubectl get pods -l serving.knative.dev/configuration=event-display -o jsonpath='{.items[0].metadata.name}')

~$ kubectl logs $POD
Defaulted container "user-container" out of: user-container, queue-proxy
☁️  cloudevents.Event
Context Attributes,
  specversion: 1.0
  type: com.icloud2native.flow.seq
  source: Curl
  id: 100001
Data,
   {"message":"Hello from Knative Eventing KafkaChannel"}