定西做网站,一个网站做数据分析要多少钱,wordpress插件自动,湖南建筑一体化平台kubernetes集群 应用实践 kafka部署 零.1、环境说明 零.2、kafka架构说明 zookeeper在kafka集群中的作用 一、Broker注册 二、Topic注册 三、Topic Partition选主 四、生产者负载均衡 五、消费者负载均衡 一、持久化存储资源准备 1.1 创建共享目录
[rootnfsserver ~]# mkdir -…kubernetes集群 应用实践 kafka部署 零.1、环境说明 零.2、kafka架构说明 zookeeper在kafka集群中的作用 一、Broker注册 二、Topic注册 三、Topic Partition选主 四、生产者负载均衡 五、消费者负载均衡 一、持久化存储资源准备 1.1 创建共享目录
[rootnfsserver ~]# mkdir -p /vdc/kafka/data1
[rootnfsserver ~]# mkdir -p /vdc/kafka/data2
[rootnfsserver ~]# mkdir -p /vdc/kafka/data3 1.2 验证共享目录
[rootnfsserver ~]# tree /vdc/kafka
/vdc/kafka
├── data1
├── data2
└── data3
3 directories, 0 files 1.3 共享存储目录
[rootnfsserver ~]# 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) [rootnfsserver ~]# exportfs -a [rootnfsserver ~]# 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资源清单文件
[rootnginxk8syaml kafka]# cat kafka-pv.yaml
apiVersion: v1
kind: PersistentVolume
metadata:name: kafka-data1
spec:capacity:storage: 3GiaccessModes:- ReadWriteOncenfs:server: nfs.wego.redpath: /vdc/kafka/data1
---
apiVersion: v1
kind: PersistentVolume
metadata:name: kafka-data2
spec:capacity:storage: 3GiaccessModes:- ReadWriteOncenfs:server: nfs.wego.redpath: /vdc/kafka/data2
---
apiVersion: v1
kind: PersistentVolume
metadata:name: kafka-data3
spec:capacity:storage: 3GiaccessModes:- ReadWriteOncenfs:server: nfs.wego.redpath: /vdc/kafka/data3 2.2 定位资源清单文件位置 2.3 应用资源清单文件
[rootmaster1 ~]# 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创建情况
[rootmaster1 ~]# 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资源清单文件
[rootnginxk8syaml kafka]# cat kafka.yaml
---
apiVersion: v1
kind: Service
metadata:name: kafka-hsnamespace: defaultlabels:app: kafka
spec:ports:- port: 9093name: serverclusterIP: Noneselector:app: kafka
---
apiVersion: policy/v1beta1
kind: PodDisruptionBudget
metadata:name: kafka-pdbnamespace: default
spec:selector:matchLabels:app: kafkamaxUnavailable: 1
---
apiVersion: apps/v1
kind: StatefulSet
metadata:name: kafkanamespace: default
spec:serviceName: kafka-hsreplicas: 3selector:matchLabels:app: kafkapodManagementPolicy: ParallelupdateStrategy:type: RollingUpdatetemplate:metadata:labels:app: kafkaspec:terminationGracePeriodSeconds: 300containers:- name: k8skafkaimagePullPolicy: IfNotPresentimage: harbor.wego.red/library/kubernetes-kafka:v10.2.1resources:requests:memory: 256Micpu: 0.1ports:- containerPort: 9093name: servercommand:- sh- -c- exec kafka-server-start.sh /opt/kafka/config/server.properties --override broker.id${HOSTNAME##*-} \--override listenersPLAINTEXT://:9093 \--override zookeeper.connectzk-cs.default.svc.cluster.local:2181 \--override log.dir/var/lib/kafka \--override auto.create.topics.enabletrue \--override auto.leader.rebalance.enabletrue \--override background.threads10 \--override compression.typeproducer \--override delete.topic.enablefalse \--override leader.imbalance.check.interval.seconds300 \--override leader.imbalance.per.broker.percentage10 \--override log.flush.interval.messages9223372036854775807 \--override log.flush.offset.checkpoint.interval.ms60000 \--override log.flush.scheduler.interval.ms9223372036854775807 \--override log.retention.bytes-1 \--override log.retention.hours168 \--override log.roll.hours168 \--override log.roll.jitter.hours0 \--override log.segment.bytes1073741824 \--override log.segment.delete.delay.ms60000 \--override message.max.bytes1000012 \--override min.insync.replicas1 \--override num.io.threads8 \--override num.network.threads3 \--override num.recovery.threads.per.data.dir1 \--override num.replica.fetchers1 \--override offset.metadata.max.bytes4096 \--override offsets.commit.required.acks-1 \--override offsets.commit.timeout.ms5000 \--override offsets.load.buffer.size5242880 \--override offsets.retention.check.interval.ms600000 \--override offsets.retention.minutes1440 \--override offsets.topic.compression.codec0 \--override offsets.topic.num.partitions50 \--override offsets.topic.replication.factor3 \--override offsets.topic.segment.bytes104857600 \--override queued.max.requests500 \--override quota.consumer.default9223372036854775807 \--override quota.producer.default9223372036854775807 \--override replica.fetch.min.bytes1 \--override replica.fetch.wait.max.ms500 \--override replica.high.watermark.checkpoint.interval.ms5000 \--override replica.lag.time.max.ms10000 \--override replica.socket.receive.buffer.bytes65536 \--override replica.socket.timeout.ms30000 \--override request.timeout.ms30000 \--override socket.receive.buffer.bytes102400 \--override socket.request.max.bytes104857600 \--override socket.send.buffer.bytes102400 \--override unclean.leader.election.enabletrue \--override zookeeper.session.timeout.ms6000 \--override zookeeper.set.aclfalse \--override broker.id.generation.enabletrue \--override connections.max.idle.ms600000 \--override controlled.shutdown.enabletrue \--override controlled.shutdown.max.retries3 \--override controlled.shutdown.retry.backoff.ms5000 \--override controller.socket.timeout.ms30000 \--override default.replication.factor1 \--override fetch.purgatory.purge.interval.requests1000 \--override group.max.session.timeout.ms300000 \--override group.min.session.timeout.ms6000 \--override inter.broker.protocol.version0.10.2-IV0 \--override log.cleaner.backoff.ms15000 \--override log.cleaner.dedupe.buffer.size134217728 \--override log.cleaner.delete.retention.ms86400000 \--override log.cleaner.enabletrue \--override log.cleaner.io.buffer.load.factor0.9 \--override log.cleaner.io.buffer.size524288 \--override log.cleaner.io.max.bytes.per.second1.7976931348623157E308 \--override log.cleaner.min.cleanable.ratio0.5 \--override log.cleaner.min.compaction.lag.ms0 \--override log.cleaner.threads1 \--override log.cleanup.policydelete \--override log.index.interval.bytes4096 \--override log.index.size.max.bytes10485760 \--override log.message.timestamp.difference.max.ms9223372036854775807 \--override log.message.timestamp.typeCreateTime \--override log.preallocatefalse \--override log.retention.check.interval.ms300000 \--override max.connections.per.ip2147483647 \--override num.partitions3 \--override producer.purgatory.purge.interval.requests1000 \--override replica.fetch.backoff.ms1000 \--override replica.fetch.max.bytes1048576 \--override replica.fetch.response.max.bytes10485760 \--override reserved.broker.max.id1000 env:- name: KAFKA_HEAP_OPTSvalue : -Xmx256M -Xms256M- name: KAFKA_OPTSvalue: -Dlogging.levelINFOvolumeMounts:- name: datadirmountPath: /var/lib/kafkareadinessProbe:exec:command:- sh- -c- /opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-serverlocalhost:9093volumeClaimTemplates:- metadata:name: datadirspec:accessModes: [ ReadWriteOnce ]resources:requests:storage: 3Gi 3.1.1 修改容器镜像 在课程目录中有相关镜像直接导入harbor主机并上传到harbor仓库 修改容器镜像
containers:- name: k8skafkaimagePullPolicy: IfNotPresentimage: 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 listenersPLAINTEXT://:9093 \--override zookeeper.connectzk-cs.default.svc.cluster.local:2181 \ 3.2 应用kafka资源清单文件
[rootmaster1 ~]# 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创建情况
[rootmaster1 ~]# 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 [rootmaster1 ~]# 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 [rootmaster1 ~]# kubectl exec -it zok-1 -n default bash
rootzok-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
[rootmaster1 ~]# 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 87m5.2 进入kafka pod
[rootmaster1 ~]# 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 rootkafka-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 rootkafka-0:/opt/kafka/bin# ./kafka-topics.sh --zookeeper zk-cs.default.svc.cluster.local:2181 --list 输出
test 5.5 模拟生产者 终端1 rootkafka-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 worldCTRLC即可退出 5.6 模拟消费者 终端2 [rootmaster1 ~]# kubectl exec -it kafka-0 bash
rootkafka-0:/# cd /opt/kafka/bin
rootkafka-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:9093kafka-1.kafka-hs.default.svc.cluster.local:9093kafka-2.kafka-hs.default.svc.cluster.local:9093
