简介kafka-example.rar 是一份基于 Java 的 Apache Kafka 示例工程面向需要了解 Kafka 核心机制并将消息队列引入 Web 服务场景的开发者定位在入门到中级实战之间。压缩包共 30 个文件、6.95MB以 19 个 jar 依赖库为主涵盖 Kafka、Zookeeper、日志与压缩编解码组件同时提供 4 个 .java 源码、4 个 .class 编译文件及 Eclipse 工程配置导入 IDE 后即可查看依赖关系并直接运行。全包已有 108 人学习下载。示例代码覆盖 Topic、Partition、Consumer Group 等核心概念的 Java 实现通过 Producer API 与 Consumer API 演示消息发布、订阅、偏移量管理和序列化器配置并能对照源码排查连接参数与消费组分配细节理解 Kafka 与 Web 服务器结合时的日志聚合和异步解耦路径通过阅读源码还可掌握生产者、消费者、分区提交和消费者组重平衡的编码方式并借助包内依赖搭建本地 Zookeeper Kafka 环境进行全流程验证。对于希望将 Kafka 集成进 Java Web 服务、实现日志实时采集、API 异步消息通信或事件驱动微服务的中级开发者这份资源提供了一个可直接修改的运行模板既能作为本地环境的学习样例也能作为后续流处理基础脚手架有效降低 Kafka 的入门门槛和排错时间。1. kafka-example.rar把 Kafka 和 Java Web 后端串起来的最小可运行工程解压这个包里面躺着 kafka-0.7.1.jar 和一整套 eclipse 工程配置.classpath、.project、.settings第一反应可能是“这版本也太老了”。但别急着删。这份 kafka-example 资源的价值不在于新而在于它把 Kafka 的核心链路——生产者发消息、消费者收消息、ZooKeeper 协调、序列化与压缩选型——用最小可运行的 Java 工程串了一遍。Web 服务器接入 Kafka无论是做访问日志聚合、API 异步消息队列还是事件驱动微服务架构最后都要落到这套 producer/consumer API 上。它的定位是 Kafka 入门脚手架新手按 src 目录里的代码跑通一遍能比看文档快十倍地理解 Kafka 原理熟手拿它对照当前 kafka-clients 的 API 演进能定位很多线上兼容性问题的根源。不需要找什么“简化版”这份资源本身就是。2. 先读依赖清单Kafka 0.7 的架构骨架和版本坐标系2.1 src 与 jar 包三组依赖对应三个用途拆开这份压缩包能还原出 Kafka 0.7.1 时代的完整工程依赖二十多个 jar 包看着唬人其实只分三组。jar 包角色kafka-0.7.1.jarKafka 核心库producer/consumer 和 broker 逻辑都在里面zookeeper-3.3.4.jar、zkclient-0.1.jar协调与分布式一致性客户端log4j-1.2.15.jarKafka 内部日志输出框架snappy-java-1.0.4.1.jar消息压缩支持Snappy 算法commons-cli-1.2.jar、jopt-simple-3.2.jarbin 脚本的命令行参数解析commons-io、commons-lang、commons-collections工具类库junit-4.1.jar、easymock、objenesis、cglib单元测试与 Mock 库scalatest-1.2.jarScala 测试支持Kafka 0.7 本身由 Scala 编写apache-rat-0.8 / rat-tasks / rat-coreASF 许可检查构建插件这三组分别对应第一组kafka、zkclient、zookeeper、log4j、snappy是运行期真正会碰到的核心第二组commons、jopt、cli是脚本和工具链支撑第三组是开发构建期的测试与检查工具。你改代码时重点看第一组即可别被那堆 easymock、cglib 干扰判断。版本坐标系才是最关键的。Kafka 0.7.1 的消息协议和消息格式与后面 0.8、0.10、2.x、3.x 完全不兼容。用这里的 jar 包去连接新版 kafka broker会在握手阶段报出 invalid protocol 一类异常。判断一个 kafka-example 能不能直接用第一件事就是统一三处版本客户端 jar、broker 服务端、ZooKeeper 客户端。三处不一致时优先以 broker 版本为准。2.2 主题、分区、副本一条消息在 Kafka 里的完整旅程初学者最容易把 Kafka 和普通 MQRabbitMQ、ActiveMQ混为一谈最本质的差别在分区模型上。主题Topic是数据分类的入口类似数据库表。生产者把消息写到某个主题下消费者按主题订阅。一个主题在物理上会被切分成多个分区Partition分区才是真正存储消息的单元每条消息落盘时归属于唯一一个分区。分区数量决定了一个主题的最大并行度同一个分区内的消息严格有序跨分区则没有全局顺序保证。分区副本Replica解决的是容错每个分区有 leader 和 follower 副本读写都走 leaderfollower 异步拉取并同步。leader 挂掉后ZK 会在 follower 里选出新 leader。生产者的 ack 配置决定了数据要多可靠acks0 发完即走acks1 等 leader 写盘acksall 等所有副本同步完。Web 场景里日志聚合可以用 acks1订单类消息建议把副本因子设到 2 到 3 且 acksall。消息写入流程在 0.7 里是这样Producer 指定 topic根据 key 的 hash 或轮询策略选择分区然后发送到该分区 leader 所在的 broker 节点leader 落盘后根据副本配置同步给 follower再向 Producer 返回 ack。消费者则从 ZK 拉取该 topic 的分区列表、自己的消费组状态按 offset 顺序消费。offset 是分区的消息位移。旧版 Kafka 把 offset 存在 ZK 中所以 zookeeper 挂掉消费者基本没法工作新版改为存在 broker 内部主题 __consumer_offsets 里。如果你在踩坑时总看到 ZK 过载直接把 offset 存储换掉是最有效的解决手段。2.3 生产者、消费者、消费者组Java API 对应的三类角色工程里 src/kafka 目录下的示例代码大概率是生产者、消费者、消费者组三个角色各一个类。在 0.7.1 的 Java API 里对应关系如下生产者kafka.javaapi.producer.Producer配合 ProducerConfig 设置 broker 列表、序列化器、ack 策略。消费者kafka.javaapi.consumer.ConsumerConnector配合 ConsumerConfig 设置 ZK 地址、group.id、offset 重置策略。消费者组多个消费者进程共用同一个 group.id组内每个消费者分配到不同分区实现负载均衡一个分区在同一时间只被组内一个消费者消费。这个模型跟现代 kafka-clients 的差异很大新版消费者不再直连 ZK而是通过 broker 协调分区再均衡offset 也提交到 broker。但角色关系没变。看懂旧版代码再去读新文档会发现新 SDK 只是把协调逻辑挪了位置核心概念还是主题分区副本这一套。3. 把示例跑起来环境搭建与 Java Producer/Consumer 落地3.1 最小部署先起 ZooKeeper 再起 Broker这份资源只有客户端代码不含 broker 安装包。要跑通先在本地搭一个最小集群架构上是 ZooKeeper 单机 Broker 单机这是官方文档最常见的起步方式。Windows 环境注意点先把 JAVA_HOME 环境变量配好Kafka 0.7 依赖 Java 6/7太新的 JDK 大概率跑不起来。ZooKeeper 可以用官方发行包里的 bin/zkServer.shLinux 下或 bin/zkServer.cmdWindows 下直接起默认监听 2181 端口。然后下载与客户端同版本的 Kafka broker 包0.7.1 的二进制发行包在 Apache 归档里能找到。解压之后不用跑什么花哨的控制台脚本直接调用主类就能起 broker# 把 config 目录下的 server.properties 里这三项改成你要的值 brokerid0 port9092 zk.connectlocalhost:2181 # 启动 brokerKafka 主类为 kafka.Kafka java -cp lib/*;config kafka.Kafka config/server.properties参数说明brokerid 是 broker 的唯一标识集群里不能重复port 是 broker 对外的服务端口9092 是 Kafka 默认约定zk.connect 指向上面的 ZK 地址broker 启动时会自动去 ZK 注册自己的元数据。注意 0.7 里没有后来版本的 kafka-server-start.sh 脚本直接用 java 主类是兼容性最好的做法。启动后最好立刻验证 ZK 里能看到 broker 注册信息。用 ZK 自带客户端敲一行命令# bin/zkCli.sh 或 bin/zkCli.cmd 进入后执行 ls /brokers/ids逻辑说明如果返回 [0]说明 brokerid0 的节点已成功注册broker 与 ZK 的握手没问题。如果一直返回空说明 broker 起来后又掉线了去看 broker 的日志目录默认 /tmp/kafka-logs里的 server.log。3.2 生产者代码连接 Broker 发送消息工程 src 目录里如果只有一个示例生产者大概率长这样这是 0.7.x 的 Java API 写法与包内 jar 配套import kafka.javaapi.producer.Producer; import kafka.producer.ProducerConfig; import kafka.producer.KeyedMessage; import java.util.Properties; public class SimpleProducer { public static void main(String[] args) { Properties props new Properties(); // broker 地址列表多个用逗号分隔 props.put(metadata.broker.list, localhost:9092); // 消息 value 的序列化器字符串场景用 StringEncoder props.put(serializer.class, kafka.serializer.StringEncoder); // key 序列化器类型 props.put(key.serializer.class, kafka.serializer.StringEncoder); // acks: 0 不等待1 等 leader 写盘-1 等所有副本 props.put(request.required.acks, 1); ProducerConfig config new ProducerConfig(props); ProducerString, String producer new ProducerString, String(config); for (int i 0; i 100; i) { KeyedMessageString, String message new KeyedMessageString, String( web-access-log, // topic 192.168.1. (i % 10), // key GET /index.html i); // value producer.send(message); } producer.close(); } }逻辑说明metadata.broker.list 只要求至少写一个 broker 地址Producer 会通过它去拉取全部分区元数据serializer.class 决定 value 如何转成字节request.required.acks 控制可靠性等级。这个例子里我用 ip 当 key、访问日志当 value符合 Web 服务器日志聚合的典型数据形状。send 方法在 0.7 里默认是异步批量发送只要不主动 close 或 flush消息会攒在内存缓冲区里按批次发往 broker。参数说明中需要划个边界metadata.broker.list 里写多个 broker 时第一个不可用会自动尝试下一个但如果所有 broker 都不可达send 会抛 TimeoutException。acks1 时 leader 写盘即返回极端故障下会丢消息日志聚合场景可以接受涉及资金或订单的场景不建议。3.3 消费者代码通过 ZK 订阅并消费消息消费者端的 Java 代码0.7.x 长这样import kafka.consumer.ConsumerConfig; import kafka.consumer.ConsumerIterator; import kafka.consumer.KafkaStream; import kafka.consumer.Consumer; import kafka.javaapi.consumer.ConsumerConnector; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties; public class SimpleConsumer { public static void main(String[] args) { Properties props new Properties(); props.put(zookeeper.connect, localhost:2181); props.put(group.id, web-log-group); // smallest: 从头开始消费; largest: 只消费新消息 props.put(auto.offset.reset, smallest); props.put(auto.commit.enable, true); props.put(auto.commit.interval.ms, 1000); ConsumerConfig config new ConsumerConfig(props); ConsumerConnector connector Consumer.createJavaConsumerConnector(config); MapString, Integer topicCount new HashMapString, Integer(); // 表示给本消费者分配 1 个分区流 topicCount.put(web-access-log, 1); MapString, ListKafkaStreambyte[], byte[] streams connector.createMessageStreams(topicCount); for (KafkaStreambyte[], byte[] stream : streams.get(web-access-log)) { ConsumerIteratorbyte[], byte[] it stream.iterator(); while (it.hasNext()) { byte[] value it.next().message(); System.out.println(consume: new String(value)); } } } }逻辑说明与新版不同0.7 的消费者必须先连 ZK再由 ZK 协调分区分配。createMessageStreams 返回的是按分区拆好的流列表每个 stream 对应一个分区。auto.offset.resetsmallest 表示当 group.id 没有历史 offset 时从头消费该主题全部分区largest 则只消费启动之后的新消息。auto.commit.enabletrue 会让消费者定期把消费位移提交到 ZK避免重启后重复消费代价是每条消息处理完之后最好立刻打印或落库否则进程在自动提交窗口内崩溃会丢一小段消息。这段代码实际生产时要注意三个地方topicCount 里的数值决定了本进程拉取多少个分区流如果分区总数大于消费者实例数会有消费者分到多个分区流group.id 不同则视为不同消费组消息会被重复发给每个组相同 group.id 情况下分区只会被组内某一个消费者拿到。3.4 命令行工具验证收发全链路代码写完先别急着上工程用 bin 目录下的命令行工具做一次冒烟测试最快。0.7.1 自带的工具是 kafka-console-producer 和 kafka-console-consumer分别对应生产者与消费者。# 先启动一个终端消费 test-topic 的消息 bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test-topic --from-beginning # 另开终端发 3 条消息 bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test-topic hello kafka hello java hello web-server逻辑说明console-producer 逐行读取标准输入发消息console-consumer 连接 ZK 订阅主题并打印内容。如果消费者窗口实时打出了那三行说明 broker 的收发链路、ZK 协调、序列化转换全部正常这时候再回到 eclipse 里跑 Java 端代码报错范围就能收窄到代码本身而不是环境问题。如果消息打不出来优先怀疑三件事topic 是否真的创建了0.7 默认 auto.create.topics.enabletrue 会自动建主题但某些发行版关掉了consumer 与 producer 的 broker 地址是否一致ZK 是否仍是同一台。这三项排查完90% 的“命令行不通”都能解决。4. Kafka 与 Web 服务器结合三种典型落地方案4.1 访问日志聚合把 Tomcat/Nginx 的访问日志实时送进 KafkaWeb 服务器每天都产生大量访问日志文件落盘后交给 ELK 定时采集时效性差。Kafka 在这里充当“日志总线”Web 接入层Nginx或应用容器Tomcat通过 log4j appender 或自研采集线程把 access log 实时写进 Kafka 主题下游由实时分析、离线数仓、告警规则三个消费组各取所需。工程示例里用 StringEncoder 发的消息正好映射一行日志。实际落地一般用 log4j 的 KafkaAppender但 0.7 时代没有官方 appender常见做法是自己写一个 log4j Appender在 append 方法里调用 Producer 的 sendpublic class KafkaLogAppender extends AppenderSkeleton { private ProducerString, String producer; private String topic; private String brokerList; protected void append(LoggingEvent event) { // 把日志事件渲染成一行发送到指定 topic String line event.getRenderedMessage(); producer.send(new KeyedMessageString, String( topic, event.getThreadName(), line)); } }逻辑说明这个 appender 的关键设计点是 key 用线程名而不用随机值能让同一线程产生的日志落到同一分区保证单请求的日志顺序。brokerList 和 topic 由 log4j.properties 注入。生产上要控制缓冲Kafka 生产者本身是异步批量发送给 appender 加一个独立队列日志量峰值时不阻塞业务线程是最稳妥的做法。参数说明这个方案里真正要紧的是 producer 的 batch 参数和业务线程解耦。我的习惯是把 Producer 做成静态单例不要每条日志 new 一个否则高并发下 ZK 连接和 TCP 连接会被打爆。一个 8C16G 的 Web 节点日志峰值一般能稳定跑到 50MB/s 级瓶颈往往在 snappy 压缩而不是网络。4.2 API 消息队列Web 服务间异步解耦第二个典型场景是让 Kafka 充当 Web 服务之间的异步消息中间件。比如下单接口收到请求后并不直接调用库存服务和优惠券服务而是把“订单创建成功”事件写入 Kafka由下游服务各自消费。这样下单接口的响应时间从几百毫秒降到十几毫秒下游服务挂了也不影响主链路下单。实现上你需要给每个下游服务单独建一个消费者组。因为同一主题下的不同 group.id 都会收到全量消息库存组、优惠券组各自消费同一批事件互不干扰// 库存服务group id 为 stock-service props.put(group.id, stock-service); // 优惠券服务group id 为 coupon-service props.put(group.id, coupon-service);参数说明group.id 的这个设计经常出现在相关岗位面试里——Kafka 如何保证一条消息只被消费一次答案是“同一个消费者组内保证一次”不同组则各拿一份。如果你想模拟一个传统的点对点队列就把所有消费者放进同一组想模拟发布订阅就为每个订阅方建独立组。实践时有个边界要牢记0.7 的消费者把 offset 存在 ZK消费者组扩容缩容时ZK 会触发分区再均衡。再均衡期间该组消费者会短暂停止消费如果每个消费者的处理时间都在 10 秒以上均衡风暴会反复出现。所以异步任务消费者建议把单条处理时间控制在 1 秒内否则要考虑继续拆主题或改手动提交 offset。4.3 事件驱动架构消费者角色拆分与多级缓冲再往后走一步就是事件驱动架构。把 Web 服务器内部复杂的同步调用链拆成事件流每个事件经过 Kafka 时可以被多个消费方并行处理也能被流处理引擎做窗口聚合、状态统计。在 0.7 时代没有官方流处理库你在工程里看到的 Kafka 代码其实都是最底层 producer/consumer。要做流式聚合常见做法是消费者拉取消息后按时间窗口在内存里做状态化计算再把聚合结果写入另一个 topic。下面的代码演示一个最简单的“每分钟访问计数”消费者// 假设 topic 里每行是访问日志value 包含访问时间和 url props.put(group.id, pv-counter); MapString, Long countByMinute new HashMapString, Long(); while (it.hasNext()) { String line new String(it.next().message()); String minute line.substring(0, 16); // yyyy-MM-dd HH:mm Long c countByMinute.get(minute); countByMinute.put(minute, (c null ? 0L : c) 1L); // 到分钟边界时把统计结果发送到聚合 topic if (countByMinute.size() 100) { sendToResultTopic(countByMinute); countByMinute.clear(); } }逻辑说明这里用内存 HashMap 做窗口聚合简单但有个前提单消费者实例下的单分区流顺序能保证同一个 minute 窗口内消息不跨实例。如果主题被拆成多个分区计数必须按 key 路由到同一分区否则窗口会被拆散。上面的 substring 取分钟这个做法只对格式固定并且 key 路由正确的输入有意义生产上需要更严格的时间分桶。事件驱动这条路0.7 能走通但基础设施太旧offset 在 ZK、无事务、无幂等生产者真上生产建议至少迁移到较新的 Kafka 版本。这份资源更适合当作理解底层模型的黑盒对照而不是直接搬到生产环境。5. 避坑跑 Kafka 示例最常见的 5 个翻车现场5.1 消费者启动即报错ZooKeeper 连不上现象Consumer 启动后控制台立即抛出 org.apache.zookeeper.KeeperException$ConnectionLossException或者提示 Unable to connect to zookeeper。原因多数情况下不是代码问题而是 ZK 根本没起来或者 zookeeper.connect 写的地址端口与 ZK 实际监听不一致。0.7 消费者启动时需要先和 ZK 建立会话会话建立失败整个进程直接退出不像后面版本里有重试等待。解决先在本机测telnet localhost 2181是否能连通确认 ZK 进程存在jps 看到 QuorumPeerMain确认 zookeeper.connect 的地址没有写错成 broker 地址。还有一类隐蔽情况ZK 起来了但磁盘目录没有写权限导致会话建立后被强制关闭日志里能看到 Session closed to client这时给 ZK 数据目录加写权限并重启即可。5.2 生产者不发消息也不报错现象producer.send() 调用返回了但消费者端一条消息都收不到broker 日志里也没有该 topic 的写入记录。原因0.7 的 producer 是异步批量模式消息先攒在内存里只有在 batch 满或者刷新间隔到了才会真正发出去。如果主线程 send 之后立刻 close 或者 main 方法结束producer 还没来得及 flush消息就丢了。另一个情况是 send 抛出的异常被上层吞掉了比如打印到日志没往外抛在主流程看起来就是“没报错”。解决在 send 之后显式调用 producer.close()或者在配置中把 producer.type 设为 sync 强制同步发送牺牲吞吐但保证可观测性。同时排查时优先看 broker 日志里有没有收到 ProduceRequest而不是只盯着自己的程序。如果 send 异常被吞先改代码把 send 的返回值或异常打印出来这是最快定位手段。5.3 消费者重启后重复消费一批消息现象消费者第一次运行消费了 1 万条重启后发现同一条消息又消费了一次。原因offset 自动提交机制没有生效。0.7 的 auto.commit.enable 默认开启但提交是有间隔的默认 60 秒如果消费者在处理消息期间崩溃或主动退出最近一段时间没有提交 offset重启后就会从头消费这段没提交的消息。这不是丢数据而是 at-least-once 语义的正常表现Kafka 保证不丢但不能保证不重复。解决要避免重复消费要么把 auto.commit.interval.ms 调小比如 1000 毫秒要么在业务处理成功后再手动调用 offset 提交要么在下游做幂等消费记录带唯一 ID 去重。对日志聚合场景重复消费几行可以接受对订单通知这类场景必须手动提交加幂等表双保险。5.4 消息延迟高消费者吞吐上不去现象消息从发送到被消费延迟时不时飙到几十秒消费者 CPU 没打满但消费的条数上不去。原因一类是消费者单条处理里做太多耗时操作比如同步落库、调第三方 API导致拉取循环阻塞另一类是多分区但消费者进程只有一个分区流都挤在一个线程里再一类是网络带宽或 snappy 压缩成为瓶颈消息体积太大。解决先看消费者线程模型——一个 KafkaStream 对应一个分区默认在一个循环里迭代如果处理逻辑耗时 500ms整条消费链路就卡 500ms。典型调法是把处理逻辑放到线程池让 stream 循环只负责拉消息同时检查每个分区最大消息大小以及 batch 大小。延迟排查看消费者和生产者两端的时间差优先对齐系统时钟否则延迟数据本身也是错的。5.5 序列化器不一致导致消费端乱码现象生产者发送时使用 ByteBuffer 或对象序列化消费者用 StringDecoder 去读打印出来的全是乱码或反序列化直接抛异常。原因0.7 的 serializer 和 decoder 是客户端两端配置的Kafka 不校验消息格式。两端不匹配时broker 不会报错只有消费者读的时候才会暴露。这是最隐蔽的一类问题因为生产者看起来完全正常。解决全链路统一序列化方式最简单的做法是两端都用字符串StringEncoder StringDecoder对象数据先用 JSON 序列化成字符串再写入。如果你想发二进制数据则两端都用对应的 ByteArray 序列化器并且约定好字节序。这里没有捷径可走序列化器配置必须写进团队代码规范不然后面接入方一多各写各的排错会耗掉大量时间。6. 进阶调参把示例从“能跑”改成“扛得住”的配置清单示例代码能跑通之后下一步是把配置从 demo 档位调到接近生产档位。我一般按顺序调四组参数。第一组是生产者可靠性。request.required.acks 从 1 调到 -1等待全部副本确认并设置 min.insync.replicas 保证至少两个副本在线避免“只有一个副本也返回成功”的假象。但同时要把 producer 的 batch.num.messages 调大0.7 里默认是 200高吞吐场景调到 2000 到 5000把吞吐和可靠性平衡住。第二组是消费者提交策略。auto.commit.interval.ms 从默认 60 秒压到 1 到 5 秒如果是在线交易类任务直接把 auto.commit.enable 设为 false改由代码在业务处理成功后再手动提交。幂等消费靠的是业务里的唯一 ID 去重不靠 offset。第三组是服务端容量。broker 的 log.retention.hours 日志保留时长日志聚合场景建议按磁盘配额反推例如每台 Web 节点每天产生 20GB 日志则总保留量除以日产量就是可保留天数。增加 log.segment.bytes 能减少 segment 文件数量提升顺序读性能代价是清理更粗粒度。第四组是消息大小上限。默认的 message.max.bytes 只有 1MB 左右如果要传图片元数据或者大 JSON必须先改 broker 端 message.max.bytes再对生产者设相同大小的 max.request.size消费者也一样。三处缺一都会出现“消息发不出去或消费读不全”的尴尬结果。调完参数之后建议配一个最简单的可视化方式来看 lag0.7 没有官方控制台常见做法是用 ZK 手工查消费者组的 offset也可以用 bin 下的消费脚本加 from-beginning 对比消费条数。有条件的直接把 broker 的 JMX 指标producer 发送速率、consumer 拉取速率接到监控平台。这些都比拍脑袋调参靠谱。从第一次排错到现在我每拆一个 Kafka 示例类资源都会强制走一遍类似流程先对齐版本再起单机 broker把 producer 和 consumer 各跑通最后按上面四组参数过一遍。这套流程救过我很多次——很多问题不是你代码不行而是消息系统的旧版本惯性太大。希望帮到你。本文还有配套的精品资源点击获取