手动部署Kraft模式Kafka集群

手动部署Kraft模式kafka集群基本信息

下载安装包

kafka安装包官网下载
下载完成后上传到服务器/opt目录下解压

tar -xf kafka_2.12-3.5.1.tgz将文件夹重命名为kafkamv kafka_2.12-3.5.1 kafka

生成集群随机uuid

[root@iamdemo1 ~ ]cd /opt/kafka/bin[root@iamdemo1 bin]# sh kafka-storage.sh random-uuidCOY-4PxTT82yfvGKqAepwQ

配置kafka集群的kraft模式参数

参数位置：/opt/kafka/config/kraft/server.properties

##角色可同时为broker和controllerprocess.roles=broker,controller ##node.id为当前服务器作为节点的idnode.id=1  ##定义投票节点，用于选举Master,每个节点都必须配置controller.quorum.voters=1@172.29.145.157:9093,2@172.29.145.182:9093,3@172.29.145.183:9093 ##9092为每个broker的通信端口，9093为controller节点的通信端口，如果一个节点是混合节点那就需要同时监听两个端口listeners=PLAINTEXT://:9092,CONTROLLER://:9093     #,EXTERNAL://172.29.145.157:9094 ##broker内部监听协议inter.broker.listener.name=PLAINTEXT ##对外公开的端口#advertised.listeners=EXTERNAL://172.29.145.157:9094,PLAINTEXT://:9092 advertised.listeners=PLAINTEXT://172.29.145.157:9092controller.listener.names=CONTROLLER listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL,EXTERNAL:PLAINTEXTnum.network.threads=3num.io.threads=8socket.send.buffer.bytes=102400socket.receive.buffer.bytes=102400socket.request.max.bytes=104857600##kafka数据默认存储的地方log.dirs=/opt/kafka/kraft-combined-logs ##每一个topic默认的分区num.partitions=6 ##恢复线程num.recovery.threads.per.data.dir=2 ##用于存储消费者组的消费偏移量信息的特殊主题，用于在发生故障时或者重新加入时能够恢复到之前的消费位置offsets.topic.replication.factor=3 ##用于存储事务状态信息的特殊主题，kafka支持事务性写入，当生产者使用事务模式写入数据时，信息会写入这个主题transaction.state.log.replication.factor=3transaction.state.log.min.isr=3auto.create.topics.enable=falsedefault.replication.factor=3##数据的存储时间log.retention.hours=168##设置kafka一个数据段最大值1Glog.segment.bytes=1073741824##检查数据过期时间300s一次log.retention.check.interval.ms=300000

使用集群uuid格式化kafka数据存储目录

[root@iamdemo1 bin]# sh kafka-storage.sh format -t COY-4PxTT82yfvGKqAepwQ -c ../config/kraft/server.propertiesFormatting /opt/kafka/kraft-combined-logs with metadata.version 3.5-IV2.

修改启动脚本，暴露JMX端口，配置JVM内存大小

##注意，jdk必须为64位jdk，否则JVM内存大小设置为4G会报错启动失败，查看方法java --version如果没有显示64bit的都是32位if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then    export KAFKA_HEAP_OPTS="-Xmx4G -Xms4G"    export JMX_PORT="9999"fi

修改kafka日志配置参数，避免一小时生成一个日志文件

文件位置：/opt/kafka/config/log4j.properties

##添加两条日志最大量log4j.appender.kafkaAppender.MaxFileSize=100MBlog4j.appender.kafkaAppender.MaxBackupIndex=5##修改DatePatternlog4j.appender.kafkaAppender.DatePattern='.'yyyy-MM-dd

使用脚本启动kafka集群

[root@iamdemo1 bin]# sh kafka-server-start.sh -daemon ../config/kraft/server.properties

查看集群状态

通过jps即可看到kafka进程是否在运行，也可通过ps aux查看查看topic副本分布情况可以通过./kafka-topics.sh --describe  --bootstrap-server 172.29.145.157:9092,172.29.145.182:9092,172.29.145.183:9092Topic: test-topic1      TopicId: U3-YpQ4QQqWM7jQ-8uAxjA PartitionCount: 6       ReplicationFactor: 3    Configs: segment.bytes=1073741824        Topic: test-topic1      Partition: 0    Leader: 3       Replicas: 3,1,2 Isr: 3,1,2        Topic: test-topic1      Partition: 1    Leader: 1       Replicas: 1,2,3 Isr: 1,2,3        Topic: test-topic1      Partition: 2    Leader: 2       Replicas: 2,3,1 Isr: 2,3,1        Topic: test-topic1      Partition: 3    Leader: 1       Replicas: 1,2,3 Isr: 1,2,3        Topic: test-topic1      Partition: 4    Leader: 2       Replicas: 2,3,1 Isr: 2,3,1        Topic: test-topic1      Partition: 5    Leader: 3       Replicas: 3,1,2 Isr: 3,1,2