1. 配置Kafka日志基础参数
在server.properties中设置日志存储目录（log.dirs）和日志段大小（log.segment.bytes），这是日志管理的基础。例如：
log.dirs=/home/kafka/logs # 指定日志存储路径（需提前创建目录并授权）
log.segment.bytes=1073741824 # 单个日志段最大1GB（默认值，可根据磁盘空间调整）
日志段是Kafka日志的基本单位，合理设置段大小能平衡日志分割频率与查询效率。
2. 选择合适的日志清理策略
Kafka支持两种核心清理策略，需根据业务需求选择：
log.cleanup.policy=delete # 启用删除策略
log.retention.hours=168 # 日志保留7天（168小时）
log.retention.bytes=1073741824 # 每个分区最大1GB（可选，限制单分区大小）
log.cleanup.policy=compact # 启用压缩策略
log.cleaner.enable=true # 开启压缩器（默认开启）
log.cleaner.threads=4 # 压缩线程数（根据CPU核心数调整）
3. 使用logrotate实现自动化日志轮转
通过logrotate工具定期分割、压缩Kafka日志，避免日志文件过大。创建/etc/logrotate.d/kafka配置文件，内容如下：
/home/kafka/logs/*.log {
daily # 每天轮转
missingok # 忽略缺失文件
rotate 7 # 保留7份旧日志
compress # 压缩旧日志（使用gzip）
delaycompress # 延迟压缩（避免压缩正在写入的日志）
ifempty # 空日志也轮转
notifempty # 非空日志才轮转
create 0644 kafka kafka # 创建新日志文件并设置权限
}
配置完成后，logrotate会自动执行轮转任务，无需手动干预。
4. 设置定时任务补充清理
若logrotate无法满足需求（如按大小清理），可通过crontab编写脚本定期删除旧日志。例如，每天凌晨删除7天前的日志：
# 编辑当前用户的crontab
crontab -e
添加以下内容：
0 0 * * * find /home/kafka/logs/ -type f -name "*.log" -mtime +7 -delete
此脚本会查找/home/kafka/logs/目录下修改时间超过7天的.log文件并删除，释放磁盘空间。
5. 监控日志状态与告警
通过监控工具实时跟踪日志大小、清理状态，及时发现异常：
kafka.log:type=Log,name=Size表示日志大小），设置磁盘空间阈值告警（如磁盘使用率超过80%时触发）。6. 优化日志压缩配置
若使用压缩策略，调整压缩参数能提升效率：
compression.type=snappy # 压缩算法（可选gzip/snappy/lz4/zstd，snappy平衡性能与压缩比）
log.cleaner.io.max.bytes.per.second=104857600 # 压缩带宽限制（100MB/s，避免影响生产）
log.cleaner.threads=4 # 压缩线程数（根据CPU核心数调整，如4核设置4线程）
压缩能减少存储占用，但会增加CPU负载，需根据服务器资源调整。
7. 手动清理日志（谨慎操作）
若需手动删除日志，需遵循以下步骤以避免数据丢失：
./bin/kafka-server-stop.sh/home/kafka/logs/下的.log、.index、.timeindex文件）。kafka-consumer-groups.sh工具重置消费者偏移量。./bin/kafka-server-start.sh config/server.properties注意：手动清理前务必备份重要数据，建议在测试环境验证后再操作。
