kubernetes集群应用实践 kafka部署

发布时间：2023年12月23日

kubernetes集群应用实践 kafka部署

零.1、环境说明

零.2、kafka架构说明

zookeeper在kafka集群中的作用

一、Broker注册

二、Topic注册

三、Topic Partition选主

四、生产者负载均衡

五、消费者负载均衡

一、持久化存储资源准备

1.1 创建共享目录

[root@nfsserver ~]# mkdir -p /vdc/kafka/data1
[root@nfsserver ~]# mkdir -p /vdc/kafka/data2
[root@nfsserver ~]# mkdir -p /vdc/kafka/data3

1.2 验证共享目录

[root@nfsserver ~]# tree /vdc/kafka
/vdc/kafka
├── data1
├── data2
└── data3
?
3 directories, 0 files

1.3 共享存储目录

[root@nfsserver ~]# cat /etc/exports
/vdc/zk/data1 ? *(rw,sync,no_root_squash)
/vdc/zk/data2 ? *(rw,sync,no_root_squash)
/vdc/zk/data3 ? *(rw,sync,no_root_squash)
/vdc/kafka/data1 ? ? ? ?*(rw,sync,no_root_squash)
/vdc/kafka/data2 ? ? ? ?*(rw,sync,no_root_squash)
/vdc/kafka/data3 ? ? ? ?*(rw,sync,no_root_squash)

[root@nfsserver ~]# exportfs -a

[root@nfsserver ~]# showmount -e 192.168.122.250
Export list for 192.168.122.250:
/vdc/kafka/data3 *
/vdc/kafka/data2 *
/vdc/kafka/data1 *
/vdc/zk/data3 ? ?*
/vdc/zk/data2 ? ?*
/vdc/zk/data1 ? ?*

二、k8s集群中kafka持久化存储PV准备

2.1 创建PV资源清单文件

[root@nginxk8syaml kafka]# cat kafka-pv.yaml
apiVersion: v1
kind: PersistentVolume
metadata:
 ?name: kafka-data1
spec:
 ?capacity:
 ? ?storage: 3Gi
 ?accessModes:
 ? ?- ReadWriteOnce
 ?nfs:
 ? ?server: nfs.wego.red
 ? ?path: /vdc/kafka/data1
---
apiVersion: v1
kind: PersistentVolume
metadata:
 ?name: kafka-data2
spec:
 ?capacity:
 ? ?storage: 3Gi
 ?accessModes:
 ? ?- ReadWriteOnce
 ?nfs:
 ? ?server: nfs.wego.red
 ? ?path: /vdc/kafka/data2
---
apiVersion: v1
kind: PersistentVolume
metadata:
 ?name: kafka-data3
spec:
 ?capacity:
 ? ?storage: 3Gi
 ?accessModes:
 ? ?- ReadWriteOnce
 ?nfs:
 ? ?server: nfs.wego.red
 ? ?path: /vdc/kafka/data3

2.2 定位资源清单文件位置

2.3 应用资源清单文件

[root@master1 ~]# kubectl apply -f http://nginxk8syaml.k8sonline.com.cn/kafka/kafka-pv.yaml
persistentvolume/kafka-data1 created
persistentvolume/kafka-data2 created
persistentvolume/kafka-data3 created

2.4 验证PV创建情况

[root@master1 ~]# kubectl get pv
NAME ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? CAPACITY ? ACCESS MODES ? RECLAIM POLICY ? STATUS ? ? ?CLAIM ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?STORAGECLASS ? ? ? ? ?REASON ? AGE
kafka-data1 ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?3Gi ? ? ? ?RWO ? ? ? ? ? ?Retain ? ? ? ? ? Available ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? 19s
kafka-data2 ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?3Gi ? ? ? ?RWO ? ? ? ? ? ?Retain ? ? ? ? ? Available ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? 19s
kafka-data3 ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?3Gi ? ? ? ?RWO ? ? ? ? ? ?Retain ? ? ? ? ? Available ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? 19s

三、k8s集群中部署kafka

3.1 kafka资源清单文件

[root@nginxk8syaml kafka]# cat kafka.yaml
---
apiVersion: v1
kind: Service
metadata:
  name: kafka-hs
  namespace: default
  labels:
    app: kafka
spec:
  ports:
  - port: 9093
    name: server
  clusterIP: None
  selector:
    app: kafka
---
apiVersion: policy/v1beta1
kind: PodDisruptionBudget
metadata:
  name: kafka-pdb
  namespace: default
spec:
  selector:
    matchLabels:
      app: kafka
  maxUnavailable: 1
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kafka
  namespace: default
spec:
  serviceName: kafka-hs
  replicas: 3
  selector:
    matchLabels:
      app: kafka
  podManagementPolicy: Parallel
  updateStrategy:
    type: RollingUpdate
  template:
    metadata:
      labels:
        app: kafka
    spec:
      terminationGracePeriodSeconds: 300
      containers:
      - name: k8skafka
        imagePullPolicy: IfNotPresent
        image: harbor.wego.red/library/kubernetes-kafka:v10.2.1
        resources:
          requests:
            memory: "256Mi"
            cpu: "0.1"
        ports:
        - containerPort: 9093
          name: server
        command:
        - sh
        - -c
        - "exec kafka-server-start.sh /opt/kafka/config/server.properties --override broker.id=${HOSTNAME##*-} \
          --override listeners=PLAINTEXT://:9093 \
          --override zookeeper.connect=zk-cs.default.svc.cluster.local:2181 \
          --override log.dir=/var/lib/kafka \
          --override auto.create.topics.enable=true \
          --override auto.leader.rebalance.enable=true \
          --override background.threads=10 \
          --override compression.type=producer \
          --override delete.topic.enable=false \
          --override leader.imbalance.check.interval.seconds=300 \
          --override leader.imbalance.per.broker.percentage=10 \
          --override log.flush.interval.messages=9223372036854775807 \
          --override log.flush.offset.checkpoint.interval.ms=60000 \
          --override log.flush.scheduler.interval.ms=9223372036854775807 \
          --override log.retention.bytes=-1 \
          --override log.retention.hours=168 \
          --override log.roll.hours=168 \
          --override log.roll.jitter.hours=0 \
          --override log.segment.bytes=1073741824 \
          --override log.segment.delete.delay.ms=60000 \
          --override message.max.bytes=1000012 \
          --override min.insync.replicas=1 \
          --override num.io.threads=8 \
          --override num.network.threads=3 \
          --override num.recovery.threads.per.data.dir=1 \
          --override num.replica.fetchers=1 \
          --override offset.metadata.max.bytes=4096 \
          --override offsets.commit.required.acks=-1 \
          --override offsets.commit.timeout.ms=5000 \
          --override offsets.load.buffer.size=5242880 \
          --override offsets.retention.check.interval.ms=600000 \
          --override offsets.retention.minutes=1440 \
          --override offsets.topic.compression.codec=0 \
          --override offsets.topic.num.partitions=50 \
          --override offsets.topic.replication.factor=3 \
          --override offsets.topic.segment.bytes=104857600 \
          --override queued.max.requests=500 \
          --override quota.consumer.default=9223372036854775807 \
          --override quota.producer.default=9223372036854775807 \
          --override replica.fetch.min.bytes=1 \
          --override replica.fetch.wait.max.ms=500 \
          --override replica.high.watermark.checkpoint.interval.ms=5000 \
          --override replica.lag.time.max.ms=10000 \
          --override replica.socket.receive.buffer.bytes=65536 \
          --override replica.socket.timeout.ms=30000 \
          --override request.timeout.ms=30000 \
          --override socket.receive.buffer.bytes=102400 \
          --override socket.request.max.bytes=104857600 \
          --override socket.send.buffer.bytes=102400 \
          --override unclean.leader.election.enable=true \
          --override zookeeper.session.timeout.ms=6000 \
          --override zookeeper.set.acl=false \
          --override broker.id.generation.enable=true \
          --override connections.max.idle.ms=600000 \
          --override controlled.shutdown.enable=true \
          --override controlled.shutdown.max.retries=3 \
          --override controlled.shutdown.retry.backoff.ms=5000 \
          --override controller.socket.timeout.ms=30000 \
          --override default.replication.factor=1 \
          --override fetch.purgatory.purge.interval.requests=1000 \
          --override group.max.session.timeout.ms=300000 \
          --override group.min.session.timeout.ms=6000 \
          --override inter.broker.protocol.version=0.10.2-IV0 \
          --override log.cleaner.backoff.ms=15000 \
          --override log.cleaner.dedupe.buffer.size=134217728 \
          --override log.cleaner.delete.retention.ms=86400000 \
          --override log.cleaner.enable=true \
          --override log.cleaner.io.buffer.load.factor=0.9 \
          --override log.cleaner.io.buffer.size=524288 \
          --override log.cleaner.io.max.bytes.per.second=1.7976931348623157E308 \
          --override log.cleaner.min.cleanable.ratio=0.5 \
          --override log.cleaner.min.compaction.lag.ms=0 \
          --override log.cleaner.threads=1 \
          --override log.cleanup.policy=delete \
          --override log.index.interval.bytes=4096 \
          --override log.index.size.max.bytes=10485760 \
          --override log.message.timestamp.difference.max.ms=9223372036854775807 \
          --override log.message.timestamp.type=CreateTime \
          --override log.preallocate=false \
          --override log.retention.check.interval.ms=300000 \
          --override max.connections.per.ip=2147483647 \
          --override num.partitions=3 \
          --override producer.purgatory.purge.interval.requests=1000 \
          --override replica.fetch.backoff.ms=1000 \
          --override replica.fetch.max.bytes=1048576 \
          --override replica.fetch.response.max.bytes=10485760 \
          --override reserved.broker.max.id=1000 "
        env:
        - name: KAFKA_HEAP_OPTS
          value : "-Xmx256M -Xms256M"
        - name: KAFKA_OPTS
          value: "-Dlogging.level=INFO"
        volumeMounts:
        - name: datadir
          mountPath: /var/lib/kafka
        readinessProbe:
          exec:
           command:
            - sh
            - -c
            - "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server=localhost:9093"
  volumeClaimTemplates:
  - metadata:
      name: datadir
    spec:
      accessModes: [ "ReadWriteOnce" ]
      resources:
        requests:
          storage: 3Gi

3.1.1 修改容器镜像

在课程目录中有相关镜像，直接导入harbor主机并上传到harbor仓库

修改容器镜像
containers:
      - name: k8skafka
        imagePullPolicy: IfNotPresent
        image: harbor.wego.red/library/kubernetes-kafka:v10.2.1

3.1.2 修改zookeeper连接地址

修改zookeeper.connect连接地址，使用k8s集群内域名，需要DNS服务。
command:
        - sh
        - -c
        - "exec kafka-server-start.sh /opt/kafka/config/server.properties --override broker.id=${HOSTNAME##*-} \
          --override listeners=PLAINTEXT://:9093 \
          --override zookeeper.connect=zk-cs.default.svc.cluster.local:2181 \

3.2 应用kafka资源清单文件

[root@master1 ~]# kubectl apply  -f http://nginx.wego.red/kafka/kafka.yaml
service/kafka-hs created
poddisruptionbudget.policy/kafka-pdb created
statefulset.apps/kafka created

3.3 验证kafka创建情况

[root@master1 ~]# kubectl get pods
NAME                                      READY   STATUS    RESTARTS   AGE
busybox-pod                               1/1     Running   397        28d
kafka-0                                   1/1     Running   0          106s
kafka-1                                   1/1     Running   0          106s
kafka-2                                   1/1     Running   0          106s
nfs-client-provisioner-5786f95795-54v4s   1/1     Running   4          9d
zok-0                                     1/1     Running   2          3h38m
zok-1                                     1/1     Running   1          3h38m
zok-2                                     1/1     Running   1          3h37m

四、通过zookeeper查看broker

[root@master1 ~]# kubectl get pods
NAME                                      READY   STATUS    RESTARTS   AGE
zok-0                                     1/1     Running   2          5h4m
zok-1                                     1/1     Running   1          5h3m
zok-2                                     1/1     Running   1          5h3m

[root@master1 ~]# kubectl exec -it zok-1 -n default bash
root@zok-1:/# zkCli.sh
Connecting to localhost:2181
Welcome to ZooKeeper!

[zk: localhost:2181(CONNECTING) 0] ls /
[cluster, controller, controller_epoch, brokers, zookeeper, admin, isr_change_notification, consumers, hello, config]

[zk: localhost:2181(CONNECTED) 1] ls /brokers
[ids, topics, seqid]
[zk: localhost:2181(CONNECTED) 2] ls /brokers/ids
[0, 1, 2]
[zk: localhost:2181(CONNECTED) 3] get /brokers/ids/0
{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://kafka-0.kafka-hs.default.svc.cluster.local:9093"],"jmx_port":-1,"host":"kafka-0.kafka-hs.default.svc.cluster.local","timestamp":"1581591232561","port":9093,"version":4}
cZxid = 0x20000001a
ctime = Thu Feb 13 10:53:52 UTC 2020
mZxid = 0x20000001a
mtime = Thu Feb 13 10:53:52 UTC 2020
pZxid = 0x20000001a
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x1703e20e92c0000
dataLength = 254
numChildren = 0
[zk: localhost:2181(CONNECTED) 4] get /brokers/ids/1
{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://kafka-1.kafka-hs.default.svc.cluster.local:9093"],"jmx_port":-1,"host":"kafka-1.kafka-hs.default.svc.cluster.local","timestamp":"1581591238213","port":9093,"version":4}
cZxid = 0x20000001e
ctime = Thu Feb 13 10:53:58 UTC 2020
mZxid = 0x20000001e
mtime = Thu Feb 13 10:53:58 UTC 2020
pZxid = 0x20000001e
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x1703e20e92c0001
dataLength = 254
numChildren = 0
[zk: localhost:2181(CONNECTED) 5] get /brokers/ids/2
{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://kafka-2.kafka-hs.default.svc.cluster.local:9093"],"jmx_port":-1,"host":"kafka-2.kafka-hs.default.svc.cluster.local","timestamp":"1581591242262","port":9093,"version":4}
cZxid = 0x200000022
ctime = Thu Feb 13 10:54:02 UTC 2020
mZxid = 0x200000022
mtime = Thu Feb 13 10:54:02 UTC 2020
pZxid = 0x200000022
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x2703e207a150000
dataLength = 254
numChildren = 0

五、kafka应用操作测试

5.1 查看pod

[root@master1 ~]# kubectl get pods
NAME                                      READY   STATUS    RESTARTS   AGE
busybox-pod                               1/1     Running   398        28d
kafka-0                                   1/1     Running   0          87m
kafka-1                                   1/1     Running   0          87m
kafka-2                                   1/1     Running   0          87m

5.2 进入kafka pod

[root@master1 ~]# kubectl exec -it kafka-0 -n default bash
# pwd
/
# ls
KEYS  boot  etc   lib    media  opt   root  sbin  sys  usr
bin   dev   home  lib64  mnt    proc  run   srv   tmp  var
# cd /opt/kafka/bin
# pwd
/opt/kafka/bin
# ls
connect-distributed.sh               kafka-replica-verification.sh
connect-standalone.sh                kafka-run-class.sh
kafka-acls.sh                        kafka-server-start.sh
kafka-broker-api-versions.sh         kafka-server-stop.sh
kafka-configs.sh                     kafka-simple-consumer-shell.sh
kafka-console-consumer.sh            kafka-streams-application-reset.sh
kafka-console-producer.sh            kafka-topics.sh
kafka-consumer-groups.sh             kafka-verifiable-consumer.sh
kafka-consumer-offset-checker.sh     kafka-verifiable-producer.sh
kafka-consumer-perf-test.sh          windows
kafka-mirror-maker.sh                zookeeper-security-migration.sh
kafka-preferred-replica-election.sh  zookeeper-server-start.sh
kafka-producer-perf-test.sh          zookeeper-server-stop.sh
kafka-reassign-partitions.sh         zookeeper-shell.sh
kafka-replay-log-producer.sh

5.3 创建test topic

root@kafka-0:/opt/kafka/bin# ./kafka-topics.sh --create --topic test --zookeeper zk-cs.default.svc.cluster.local:2181    --partitions 3 --replication-factor 3

输出
Created topic "test".

5.4 查看 topic

root@kafka-0:/opt/kafka/bin# ./kafka-topics.sh --zookeeper zk-cs.default.svc.cluster.local:2181 --list

输出
test

5.5 模拟生产者

终端1

root@kafka-0:/opt/kafka/bin# ./kafka-console-producer.sh --topic test --broker-list kafka-0.kafka-hs.default.svc.cluster.local:9093,kafka-1.kafka-hs.default.svc.cluster.local:9093,kafka-2.kafka-hs.default.svc.cluster.local:9093
this is a test message
hello world

CTRL+C即可退出

5.6 模拟消费者

终端2

[root@master1 ~]# kubectl exec -it kafka-0 bash
root@kafka-0:/# cd /opt/kafka/bin
root@kafka-0:/opt/kafka/bin# ./kafka-console-consumer.sh --topic test --zookeeper zk-cs.default.svc.cluster.local:2181 --from-beginning
Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].
this is a test message
hello world

六、集群内业务系统访问kafka

broker-list:

kafka-0.kafka-hs.default.svc.cluster.local:9093

kafka-1.kafka-hs.default.svc.cluster.local:9093

kafka-2.kafka-hs.default.svc.cluster.local:9093

文章来源:https://blog.csdn.net/qq_57747969/article/details/135172415
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系我的编程经验分享网邮箱：chenni525@qq.com进行投诉反馈，一经查实，立即删除！

kubernetes集群 应用实践 kafka部署

kubernetes集群 应用实践 kafka部署

零.1、环境说明

零.2、kafka架构说明

一、持久化存储资源准备

1.1 创建共享目录

1.2 验证共享目录

1.3 共享存储目录

二、k8s集群中kafka持久化存储PV准备

2.1 创建PV资源清单文件

2.2 定位资源清单文件位置

2.3 应用资源清单文件

2.4 验证PV创建情况

三、k8s集群中部署kafka

3.1 kafka资源清单文件

3.1.1 修改容器镜像

3.1.2 修改zookeeper连接地址

3.2 应用kafka资源清单文件

3.3 验证kafka创建情况

四、通过zookeeper查看broker

五、kafka应用操作测试

5.1 查看pod

5.2 进入kafka pod

5.3 创建test topic

5.4 查看 topic

5.5 模拟生产者

5.6 模拟消费者

六、集群内业务系统访问kafka

kubernetes集群应用实践 kafka部署

kubernetes集群应用实践 kafka部署