1. Kafka简介与核心概念Kafka是由Apache软件基金会开发的一个分布式流处理平台最初由LinkedIn开发并开源。它被设计用来处理高吞吐量的实时数据流具有水平扩展、容错和持久化等特性。在实际应用中Kafka通常被用作消息队列、事件溯源系统或流处理平台。Kafka的核心架构包含几个关键组件BrokerKafka集群中的每个服务器节点称为Broker负责消息的存储和转发Topic消息的类别或主题生产者将消息发送到特定Topic消费者从Topic订阅消息Partition每个Topic可以分为多个Partition实现数据的分布式存储和并行处理Producer向Kafka Topic发送消息的客户端Consumer从Kafka Topic读取消息的客户端ZookeeperKafka依赖的协调服务用于管理集群元数据和Broker状态1.1 Kafka的应用场景Kafka在以下场景中表现出色实时数据处理如用户行为分析、点击流分析日志聚合集中收集各服务的日志数据事件溯源记录系统状态变化的历史消息队列解耦生产者和消费者系统流处理结合Kafka Streams或Flink等流处理框架提示虽然Kafka功能强大但对于简单的消息队列需求如果吞吐量要求不高可以考虑更轻量级的解决方案如RabbitMQ。2. Kafka环境安装与配置2.1 系统要求与准备工作在安装Kafka前需要确保系统满足以下要求至少4GB内存生产环境建议8GB以上至少10GB磁盘空间根据数据保留策略调整Java 8或更高版本推荐OpenJDK 11网络端口9092Kafka和2181Zookeeper可用2.1.1 Java环境安装Kafka运行依赖Java环境首先检查Java是否已安装java -version如果未安装可以使用以下命令安装OpenJDK以Ubuntu为例sudo apt update sudo apt install openjdk-11-jdk2.2 Kafka安装步骤2.2.1 下载Kafka从Apache官网下载最新稳定版Kafka当前最新为3.6.0wget https://downloads.apache.org/kafka/3.6.0/kafka_2.13-3.6.0.tgz tar -xzf kafka_2.13-3.6.0.tgz cd kafka_2.13-3.6.02.2.2 启动ZookeeperKafka依赖Zookeeper进行集群协调。Kafka包中已包含Zookeeper可以快速启动bin/zookeeper-server-start.sh config/zookeeper.properties注意生产环境建议使用独立的Zookeeper集群而不是内置的单节点Zookeeper。2.2.3 启动Kafka服务新开一个终端启动Kafka服务bin/kafka-server-start.sh config/server.properties2.2.4 创建系统服务可选为了方便管理可以将Zookeeper和Kafka配置为系统服务创建Zookeeper服务文件/etc/systemd/system/zookeeper.service[Unit] DescriptionApache Zookeeper Server Afternetwork.target [Service] Typesimple ExecStart/path/to/kafka/bin/zookeeper-server-start.sh /path/to/kafka/config/zookeeper.properties ExecStop/path/to/kafka/bin/zookeeper-server-stop.sh Restarton-failure [Install] WantedBymulti-user.target创建Kafka服务文件/etc/systemd/system/kafka.service[Unit] DescriptionApache Kafka Server Afternetwork.target zookeeper.service [Service] Typesimple ExecStart/path/to/kafka/bin/kafka-server-start.sh /path/to/kafka/config/server.properties ExecStop/path/to/kafka/bin/kafka-server-stop.sh Restarton-failure [Install] WantedBymulti-user.target启用并启动服务sudo systemctl daemon-reload sudo systemctl start zookeeper sudo systemctl start kafka sudo systemctl enable zookeeper sudo systemctl enable kafka2.3 基础配置调整编辑config/server.properties文件修改以下关键配置# Broker唯一标识 broker.id0 # 监听地址 listenersPLAINTEXT://:9092 # 日志存储目录 log.dirs/tmp/kafka-logs # 默认分区数 num.partitions3 # Zookeeper连接地址 zookeeper.connectlocalhost:21813. Kafka基础操作与验证3.1 Topic管理3.1.1 创建Topic创建一个名为test的Topic1个分区1个副本bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test3.1.2 查看Topic列表bin/kafka-topics.sh --list --bootstrap-server localhost:90923.1.3 查看Topic详情bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic test3.2 生产者和消费者测试3.2.1 启动控制台生产者bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test3.2.2 启动控制台消费者新开一个终端启动消费者bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning现在可以在生产者终端输入消息在消费者终端会实时显示收到的消息。4. Java客户端开发入门4.1 项目准备创建Maven项目添加Kafka客户端依赖dependency groupIdorg.apache.kafka/groupId artifactIdkafka-clients/artifactId version3.6.0/version /dependency4.2 生产者示例import org.apache.kafka.clients.producer.*; import java.util.Properties; public class SimpleProducer { public static void main(String[] args) { // 1. 配置生产者参数 Properties props new Properties(); props.put(bootstrap.servers, localhost:9092); props.put(key.serializer, org.apache.kafka.common.serialization.StringSerializer); props.put(value.serializer, org.apache.kafka.common.serialization.StringSerializer); // 2. 创建生产者实例 ProducerString, String producer new KafkaProducer(props); // 3. 发送消息 for (int i 0; i 10; i) { ProducerRecordString, String record new ProducerRecord(test, key- i, value- i); producer.send(record, (metadata, exception) - { if (exception ! null) { exception.printStackTrace(); } else { System.out.printf(消息发送成功topic%s, partition%d, offset%d%n, metadata.topic(), metadata.partition(), metadata.offset()); } }); } // 4. 关闭生产者 producer.close(); } }4.3 消费者示例import org.apache.kafka.clients.consumer.*; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.serialization.StringDeserializer; import java.time.Duration; import java.util.Collections; import java.util.Properties; public class SimpleConsumer { public static void main(String[] args) { // 1. 配置消费者参数 Properties props new Properties(); props.put(bootstrap.servers, localhost:9092); props.put(group.id, test-group); props.put(key.deserializer, StringDeserializer.class.getName()); props.put(value.deserializer, StringDeserializer.class.getName()); props.put(auto.offset.reset, earliest); // 从最早的消息开始消费 // 2. 创建消费者实例 ConsumerString, String consumer new KafkaConsumer(props); // 3. 订阅Topic consumer.subscribe(Collections.singletonList(test)); // 4. 轮询获取消息 try { while (true) { ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(100)); for (ConsumerRecordString, String record : records) { System.out.printf(收到消息topic%s, partition%d, offset%d, key%s, value%s%n, record.topic(), record.partition(), record.offset(), record.key(), record.value()); } } } finally { // 5. 关闭消费者 consumer.close(); } } }4.4 常见配置说明生产者重要配置配置项说明推荐值acks消息确认机制all(最安全)retries发送失败重试次数3batch.size批量发送大小16384(16KB)linger.ms发送等待时间5buffer.memory缓冲区大小33554432(32MB)消费者重要配置配置项说明推荐值group.id消费者组ID必须唯一auto.offset.reset无偏移量时的策略earliest/latestenable.auto.commit自动提交偏移量true(简单场景)max.poll.records每次poll最大记录数500session.timeout.ms会话超时时间10000(10秒)5. 生产环境注意事项5.1 性能优化建议分区设计分区数应与消费者数量匹配单个分区保证有序性但会限制吞吐量一般建议每个Broker管理的分区不超过4000个消息大小Kafka适合处理中小消息1MB大消息需调整message.max.bytes和replica.fetch.max.bytes批量发送合理设置batch.size和linger.ms提高吞吐但会增加延迟需权衡5.2 监控与维护关键指标监控延迟监控生产/消费延迟吞吐量消息入/出速率存储磁盘使用率网络带宽使用率常用监控工具Kafka ManagerPrometheus GrafanaBurrow消费者延迟监控日常维护命令查看消费者组偏移量bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group test-group删除Topicbin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic test5.3 常见问题排查生产者发送失败检查网络连通性检查Broker是否正常运行检查Topic是否存在检查ACKS配置是否过高消费者无法消费检查消费者组偏移量检查分区分配情况检查auto.offset.reset配置性能瓶颈检查磁盘IO检查网络带宽检查CPU使用率检查JVM GC情况在实际项目中Kafka的配置和优化需要根据具体业务场景进行调整。建议从小规模开始逐步增加负载观察系统表现找到最适合的配置参数。