ELFK kafka 架构部署集群部署添加3台主机[rootserver6 ~]# vim /etc/hosts每台主机都要配置解析三台主机配置JDK环境[rootserver6 ~]# rpm -ivh jdk-17.0.17_linux-x64_bin.rpm[rootserver7 ~]# rpm -ivh jdk-17.0.17_linux-x64_bin.rpm[rootserver8 ~]# rpm -ivh jdk-17.0.17_linux-x64_bin.rpm三台主机安装 Kafka[rootserver6 ~]# tar zxf kafka_2.12-3.9.1.tgz -C /opt[rootserver6 ~]# mv /opt/kafka_2.12-3.9.1 /opt/kafka[rootserver6 ~]# cd /opt/kafka/config/kraft[rootserver6 kraft]# cp server.properties{,.bak} #备份配置文件[rootserver6 kraft]# mkdir /opt/kafka/logs #日志目录三台主机 Kafka 配置文件修改server6配置文件修改[rootserver6 ~]# cd /opt/kafka/config/kraft[rootserver6 kraft]# vim server.propertiesprocess.rolesbroker,controllernode.id137controller.quorum.voters137192.168.247.137:9093,138192.168.247.138:9093,139192.168.247.139:9093listenersPLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093inter.broker.listener.namePLAINTEXTadvertised.listenersPLAINTEXT://192.168.247.137:9092,CONTROLLER://192.168.247.137:9093controller.listener.namesCONTROLLERlistener.security.protocol.mapCONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSLnum.network.threads2num.io.threads2socket.send.buffer.bytes102400socket.receive.buffer.bytes102400socket.request.max.bytes104857600log.dirs/opt/kafka/logsnum.partitions3num.recovery.threads.per.data.dir1offsets.topic.replication.factor3transaction.state.log.replication.factor3transaction.state.log.min.isr1log.retention.hours168log.segment.bytes1073741824log.retention.check.interval.ms300000server7配置文件修改server8配置文件修改Kafka 集群初始化server6 节点生成储目录唯一的 UUID[rootserver6 kraft]# KAFKA_CLUSTER_ID$(/opt/kafka/bin/kafka-storage.sh random-uuid)[rootserver6 kraft]# echo ${KAFKA_CLUSTER_ID}5RipyJmTTRGgHql-85cFnQ三台主机用该 uuid 格式化 kafka 存储目录[rootserver6 kraft]# /opt/kafka/bin/kafka-storage.sh format -t 5RipyJmTTRGgHql-85cFnQ -c /opt/kafka/config/kraft/server.properties[rootserver7 kraft]# /opt/kafka/bin/kafka-storage.sh format -t 5RipyJmTTRGgHql-85cFnQ -c /opt/kafka/config/kraft/server.properties[rootserver8 kraft]# /opt/kafka/bin/kafka-storage.sh format -t 5RipyJmTTRGgHql-85cFnQ -c /opt/kafka/config/kraft/server.properties三台节点启动 Kafka[rootserver6 kraft]# /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties[rootserver6 kraft]# jps40068 Jps39847 Kafka[rootserver7 kraft]# /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties[rootserver7 kraft]# jps25150 Kafka25278 Jps[rootserver8 kraft]# /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties[rootserver8 kraft]# jps25535 Kafka25663 JpsKafka 集群可用性验证Kafka 集群任意节点创建 Topic[rootserver6 kraft]# /opt/kafka/bin/kafka-topics.sh --create --topic testtopics --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3--replication-factor 3设置主题的副本因子为 3即每个分区的副本数量为 3。副本用于数据冗余和高可用性副本数需不能超过集群中 Broker 的总数。--partitions 3设置主题的分区数为 3。分区是 Kafka 并行处理消息的基本单位影响吞吐量和扩展性。分区数一旦创建不可减少。Kafka 集群任意节点查看 Topic[rootserver7 kraft]# /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092Kafka 集群任意节点生产者测试[rootserver6 kraft]# /opt/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic testtopicsKafka 集群任意节点查看主题[rootserver8 kraft]# /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testtopics --from-beginningKafka 集群任意节点查看 topic 详细信息[rootserver7 kraft]# /opt/kafka/bin/kafka-topics.sh --bootstrap-server server6:9092,server7:9092,server8:9092 --topic testtopics --describeFIlebeat 配置文件修改[rootserver4 ~]# vim /etc/filebeat/filebeat.yml#----------------------------- Kafka output --------------------------------output.kafka:hosts: [192.168.247.137:9092, 192.168.247.138:9092, 192.168.247.139:9092]# Kafka 集群地址topic: apache-logs # 发送到的 Topic 名称partition.round_robin: # 日志按轮询方式分发到 Topic 各分区reachable_only: truerequired_acks: 1 # 至少 1 个副本确认接收平衡可靠性和性能compression: gzip # 启用压缩减少网络传输量max_message_bytes: 1048576 # 单条消息最大 1MB[rootserver4 ~]# filebeat test config -c /etc/filebeat/filebeat.yml[rootserver4 ~]# filebeat test output -c /etc/filebeat/filebeat.yml[rootserver4 ~]# systemctl restart filebeat.serviceKafka 集群任意节点消费者测试[rootserver8 kraft]# /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic apache-logs --from-beginningLogstash 消费 Topic 消息[rootserver5 ~]# cd /etc/logstash/conf.d[rootserver5 conf.d]# vim kafka.confinput {kafka {bootstrap_servers 192.168.247.137:9092,192.168.247.138:9092,192.168.247.139:9092topics [apache-logs] # 消费的 Topicgroup_id logstash-consumer-group # 消费者组 IDconsumer_threads 3 # 消费线程数 分区数最大化并行auto_offset_reset earliest # 从最早消息开始消费避免丢失codec json}}filter {grok {match { message %{HTTPD_COMBINEDLOG} }}}output {stdout {}elasticsearch {hosts [192.168.247.132:9200,192.168.247.133:9200,192.168.247.134:9200]index apachelog-%{YYYY.MM.dd}user elasticpassword westos}}[rootserver5 conf.d]# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/kafka.conf[rootserver1 ~]# ab -c1 -n 300 http://192.168.247.135/index.html[rootserver2 ~]# ab -c1 -n 200 http://192.168.247.135/index.html[rootserver3 ~]# ab -c1 -n 100 http://192.168.247.135/index.html