Kafka如何在CentOS上高效配置

一、基础环境准备

1. 安装Java环境：Kafka依赖Java 11+（推荐OpenJDK 11），通过sudo yum install java-11-openjdk-devel -y安装，验证java -version确认安装成功。
2. 下载并解压Kafka：从Apache官网下载所需版本（如3.6.1），解压至/opt/kafka目录：wget https://downloads.apache.org/kafka/3.6.1/kafka_2.13-3.6.1.tgz && tar -xzf kafka_2.13-3.6.1.tgz -C /opt/。
3. 配置防火墙：开放Kafka端口（默认9092）和Zookeeper端口（默认2181）：sudo firewall-cmd --zone public --add-port 9092/tcp --permanent && sudo firewall-cmd --zone public --add-port 2181/tcp --permanent && sudo firewall-cmd --reload。

二、Kafka核心配置优化（server.properties）

1. 基础标识与连接：设置唯一broker.id（如0）；listeners=PLAINTEXT://0.0.0.0:9092（监听所有IP的9092端口）；advertised.listeners=PLAINTEXT://your_server_ip:9092（对外公开的Broker地址，需替换为实际IP）；log.dirs=/data/kafka/logs（指定日志存储目录，建议使用独立高速磁盘）；zookeeper.connect=localhost:2181（Zookeeper连接字符串，集群环境下需列出所有节点）。
2. 性能关键参数：
   • 线程池优化：num.network.threads=3（处理网络请求的线程数，建议为CPU核数+1）；num.io.threads=8（处理磁盘I/O的线程数，建议为CPU核数的2倍）。
   • Socket缓冲区：socket.send.buffer.bytes=102400（发送缓冲区大小，默认100KB，可根据网络带宽调整至100KB-1MB）；socket.receive.buffer.bytes=102400（接收缓冲区大小）；socket.request.max.bytes=104857600（单个请求最大字节数，默认100MB，可根据消息大小调整至100MB-500MB）。
   • 分区与副本：num.partitions=3（新Topic的默认分区数，建议设置为消费者线程数的2-3倍，以提升并行处理能力）；default.replication.factor=3（副本数，建议≥2以保证数据可靠性，集群节点数≥副本数）；min.insync.replicas=2（最小同步副本数，确保数据同步的可靠性，建议≤副本数-1）。
   • 日志保留策略：log.retention.hours=168（日志保留时间，默认168小时/7天，可根据业务需求调整至72-168小时）；log.segment.bytes=1073741824（单个日志段大小，默认1GB，建议设置为1-10GB，减少日志切换频率）；log.retention.check.interval.ms=300000（日志清理检查间隔，默认5分钟，建议调整为300000ms/5分钟）。
   • 压缩配置：compression.types=lz4（启用消息压缩，推荐lz4算法，兼顾吞吐量与CPU开销，减少网络传输和磁盘IO）。

三、操作系统层优化

1. 文件系统选择：使用XFS或EXT4文件系统（XFS性能更优，推荐用于CentOS 7+），挂载时添加noatime选项（禁用文件访问时间更新，减少磁盘写操作）：mount -o noatime /dev/sdb1 /data/kafka。
2. Swap空间设置：将vm.swappiness设置为1（默认60），减少系统使用Swap的概率，避免频繁的磁盘交换影响性能：echo "vm.swappiness=1" >> /etc/sysctl.conf && sysctl -p。
3. 文件描述符限制：增加系统级和用户级的文件描述符限制，避免Kafka因打开文件过多而报错：echo "* soft nofile 65536" >> /etc/security/limits.conf && echo "* hard nofile 65536" >> /etc/security/limits.conf。
4. 内存映射限制：调整vm.max_map_count（默认65530），增加内存映射区域的数量，提升Kafka处理大量数据的能力：echo "vm.max_map_count=262144" >> /etc/sysctl.conf && sysctl -p。

四、JVM调优

1. 堆内存分配：设置-Xms（初始堆内存）和-Xmx（最大堆内存）为相同值（如4GB），避免堆内存动态调整带来的性能波动：export KAFKA_HEAP_OPTS="-Xms4G -Xmx4G"。
2. 垃圾回收器选择：使用G1GC垃圾回收器（适用于大内存场景），减少Full GC的停顿时间：export KAFKA_JVM_PERFORMANCE_OPTS="-XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35"。

五、启动与验证

1. 启动Kafka服务：使用守护进程方式启动Kafka：/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties。
2. 创建Topic测试：创建名为test的Topic（分区数3，副本数3）：/opt/kafka/bin/kafka-topics.sh --create --bootstrap-server your_server_ip:9092 --replication-factor 3 --partitions 3 --topic test。
3. 验证功能：使用控制台生产者和消费者测试消息收发：
   • 生产者：/opt/kafka/bin/kafka-console-producer.sh --bootstrap-server your_server_ip:9092 --topic test
   • 消费者：/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server your_server_ip:9092 --topic test --from-beginning。

六、监控与维护

1. 监控工具集成：使用Prometheus+Grafana监控Kafka集群的性能指标（如吞吐量、延迟、分区分布），或使用Kafka自带的kafka-exporter导出JMX指标。
2. 日志管理：定期清理Kafka日志文件（如保留7天），避免磁盘空间不足：find /data/kafka/logs -type f -name "*.log" -mtime +7 -exec rm -f {} \;。
3. 定期维护：监控Kafka集群的健康状态（如Broker是否存活、副本同步情况），及时升级Kafka版本以修复bug和提升性能。