简介这套基于Java语言的Kafka消息队列系统设计源码面向需要处理高吞吐量消息、构建实时数据管道或学习分布式中间件原理的Java开发者。项目共42个文件压缩包约77.3MB主要包含27个Java源文件、6个XML配置文件、1个YAML配置、属性文件、SQL脚本、Git忽略文件及Markdown文档等。Java文件覆盖生产者与消费者的核心逻辑、消息收发机制及Kafka集群交互XML与YAML配置文件用于设定服务器地址、数据库连接等关键参数SQL脚本可辅助初始化消息数据存储Markdown文档提供集群搭建指引方便快速理解架构并部署环境。当前已有316人学习下载。通过研读这套源码可以掌握Kafka在Java项目中的落地方式梳理负载均衡、故障转移、消息持久化等高级特性为自研消息队列或参与大数据平台开发积累直接可参考的代码与配置方案。1. Kafka的Java系统设计先搞懂它解决的三个问题再动手做Java后端绕不开消息队列选型时Kafka、RabbitMQ、RocketMQ各有拥趸但Kafka凭借高吞吐和分区模型几乎成了日志采集、流量削峰、数据同步场景的默认选项。所谓基于Java语言的Kafka消息队列系统设计源码本质是把生产端、消费端、位点、重试、幂等这些逻辑在Java工程里落地成一套可扩展的代码结构。它解决的核心问题只有三个消息怎么安全地发出去、消息怎么被正确地消费、系统出问题后怎么快速定位。适合正在做Kafka项目重构、或者打算把消息中间件从直连API升级为统一平台的团队。先说结论Kafka的运维坑远少于Java设计层的坑这套源码能否扛住生产关键在于你如何控制消费端的状态。2. Java侧的核心抽象设计从Topic到ConsumerGroup的物化Kafka本身是一个分布式的提交日志Topic是逻辑容器分区是物理并行单位ConsumerGroup是消费协调单位。Java工程落地时不能直接暴露原生客户端API给所有业务方原因有三个原生KafkaProducer和KafkaConsumer的生命周期管理容易失控每个业务方自行new实例连接数不可控各团队自行配置acks、序列化器、重试参数出问题时很难统一排查业务方直接操作ConsumerRecord会让消息体格式完全失控后续想加链路追踪、幂等标记都得改所有调用方。所以设计的第一步是在Java侧建立一层薄封装把Topic、消息体、ConsumerGroup这些概念物化成可复用的类。2.1 Producer与Consumer的接口设计为什么不该裸用原生API先看一个典型的反模式业务代码里直接new KafkaProducer发完消息不close下次调用再new一个。低并发时看不出问题一到高峰期文件描述符和内存直接被打满而且客户端没复用生产者的批量发送和缓冲机制完全失效。常见做法是封装一个MessageSender全局只持有一个KafkaProducer实例同时把序列化、发送失败补偿、监控上报都收口在这一层。// MessageSender.java 统一消息发送入口全局单例持有KafkaProducer Service public class MessageSender { private final KafkaProducerString, String producer; private final ObjectMapper mapper new ObjectMapper(); private final MessageRetryStore retryStore; // 本地失败补偿表 public MessageSender(KafkaProperties props) { Properties p new Properties(); p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, props.getBootstrapServers()); p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); p.put(ProducerConfig.ACKS_CONFIG, props.getAcks()); p.put(ProducerConfig.RETRIES_CONFIG, 3); p.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5); this.producer new KafkaProducer(p); } public void send(String topic, String key, Object body) { try { String value mapper.writeValueAsString(body); producer.send(new ProducerRecord(topic, key, value), (meta, ex) - { if (ex ! null) { log.error(kafka发送失败 topic{} key{}, topic, key, ex); retryStore.save(topic, key, value); // 入库等待定时重试 } }); } catch (JsonProcessingException e) { log.error(序列化失败 topic{}, topic, e); } } }这段代码的核心在于send方法是异步的KafkaProducer内部有缓冲区producer.send只负责把消息放入批次真正发往broker的是后台Sender线程。回调里的ex ! null代表发送失败此时把消息存入本地重试表由定时任务扫描重发而不是在回调里直接循环重试。参数说明里要注意MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION当同时设置retries 0时这个值必须大于1否则重试时会因为等待窗口为1而卡住吞吐设成5既能保留并发发送能力Kafka又保证同一个连接上最多5个未确认请求不会引发分区级乱序。2.2 消息模型与序列化设计一个可落地的消息体结构光有发送入口还不够得定消息体规范。Kafka的value可以是任意字节但生产系统里最怕的是“每个团队自己定义一套消息格式”。这套设计里我一般强制要求所有消息走同一个信封MessageEnvelope。// MessageEnvelope.java 通用消息信封 public class MessageEnvelope { private String msgId; // 全局唯一ID幂等用 private String sourceApp; // 来源系统 private long timestamp; // 生产时间 private String traceId; // 链路追踪ID private String payloadType; // 业务类型标识 private String payload; // 业务数据JSON private int retryCount; // 重试次数 }msgId用UUID还是雪花ID不重要关键是全局唯一它是消费端做幂等的唯一依据。traceId必须在入口生成并跨系统传递否则消息链路断了你根本不知道是哪一环延迟。payloadType字段存业务类型标识消费端根据它路由到对应的handleMessage方法。序列化层面payload固定用JSON字符串不直接在Kafka里传Java对象序列化后的二进制。原因有两个JSON对跨语言友好以后有Python或Go的消费者可以无缝接入Java原生序列化携带大量类元信息体积膨胀严重而且一旦类结构变更旧消息直接反序列化失败。如果后续单条消息超过几百KB、吞吐要求极高再考虑升级到Avro或Protobuf但初期用JSON能避免把设计复杂度带进消息链路。提示不要用ObjectMapper每次new一个实例它是线程安全的全局单例复用即可。序列化失败要打topic和payloadType不然线上排查不知道是哪类消息出的问题。2.3 ConsumerGroup与位点管理Java端的状态物化ConsumerGroup在Kafka里的含义是“一组协同消费订阅Topic的消费者”组内的分区分配策略决定了消息归谁。Java端设计源码时要理解Kafka保存的offset是分区的消费位点不是业务处理位点。// ConsumerConfig 手动提交位点 p.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); p.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, earliest); p.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100); p.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 180000);禁用自动提交后offset的提交完全由Java代码控制。常见做法是poll一批消息处理成功后才调用commitSync()如果处理失败不提交位点等待下次poll重新拉取。这里有一个关键参数MAX_POLL_INTERVAL_MS_CONFIG默认300秒。如果一批100条消息里有几条特别慢导致一轮处理超过300秒Kafka会认为该消费者已死亡主动触发rebalance把分区移交给其他消费者。这个设计本意是防止消费者假死但实际中经常成为“消息处理超时翻车”的元凶后面避坑章会专门讲。位点的另一个物化是重平衡监听器实现ConsumerRebalanceListener在分区被撤销或分配时记录日志、暂停或恢复业务处理确保在rebalance过程中不会出现“位点还没提交但分区已经移交”的窗口。// 注册rebalance监听器兜底提交位点 consumer.subscribe(Collections.singletonList(topic), new ConsumerRebalanceListener() { Override public void onPartitionsRevoked(CollectionTopicPartition partitions) { // 分区被收回前执行最后一次位点提交 consumer.commitSync(); } Override public void onPartitionsAssigned(CollectionTopicPartition partitions) { log.info(分区分配完成 partitions{}, partitions); } });这个监听器解决的是极端场景业务处理到一半被rebalance打断onPartitionsRevoked触发时主动提交一次把已处理但未提交的位点尽量推进缩小重复消费的窗口。注意commitSync在回调里执行会阻塞一小会儿但相比重复消费造成的业务影响这点阻塞是可接受的。3. 用Java封装Kafka的生产消费链路关键代码与参数说明上一章定的是抽象结构这一章落到可运行的生产消费链路。读者要有一个概念Kafka的Java客户端生产者是线程安全的一个KafkaProducer实例可以被多线程复用但KafkaConsumer不是线程安全的一个消费者实例只能由一个线程驱动。3.1 Producer端封装异步发送、回调与重试的配合生产者吞吐的瓶颈往往不在Kafka而在Java客户端的参数配合。下面是一组面向高吞吐的初始化配置。// ProducerInit.java 面向吞吐优先的生产者配置 Properties p new Properties(); p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 192.168.1.10:9092,192.168.1.11:9092,192.168.1.12:9092); p.put(ProducerConfig.ACKS_CONFIG, all); p.put(ProducerConfig.RETRIES_CONFIG, 3); p.put(ProducerConfig.BATCH_SIZE_CONFIG, 65536); // 64KB p.put(ProducerConfig.LINGER_MS_CONFIG, 20); p.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864); // 64MB p.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, lz4);BATCH_SIZE是单个分区的批次字节数上限默认16KB。如果单条消息只有几百字节这个值可以让多个消息凑成一个批次减少网络往返。LINGER_MS是批次等待时间默认0表示立即发送设成20毫秒意味着最多等20毫秒把更多消息攒进同一批吞吐会明显提升但消息延迟会增加20毫秒。BUFFER_MEMORY是生产者缓冲总内存默认32MB如果业务突发量很大这个值不够会导致send阻塞阻塞超过max.block.ms会抛TimeoutException。COMPRESSION_TYPE用lz4吞吐优先且CPU开销可控。这里要强调acksall配合retries3是安全性较高的组合但重试会带来重复消息所以消费端必须做幂等。enable.idempotence开启后生产者会自动处理重试产生的重复但有一个前提max.in.flight.requests.per.connection必须小于等于5且不能手动把acks配成其他值。如果业务要求消息不重不丢建议直接开幂等。重试的落地不是无限重试。我一般会在回调失败时把消息写入本地MySQL重试表用一个定时任务每30秒扫一次重试次数超过3次的置为失败状态由人工介入。这比在回调里sleep重试要稳得多因为回调线程是Kafka客户端内部的Sender线程阻塞它会影响所有消息发送。// RetryTask.java 定时补偿失败消息 Scheduled(fixedDelay 30000) public void retryFailedMessages() { ListMessageRetry retries retryStore.scan(3); for (MessageRetry r : retries) { try { producer.send(new ProducerRecord(r.getTopic(), r.getKey(), r.getValue())); retryStore.markSent(r.getId()); } catch (Exception e) { retryStore.incrRetryCount(r.getId()); } } }补偿任务的逻辑说明scan(3)只取重试次数小于3的消息发送成功就标记已处理发送失败就递增重试计数。这里要监控重试表的积压数量如果持续增长说明broker或Topic出了问题再增加重试次数只会放大故障不如直接告警。3.2 Consumer端拉取与处理poll循环的节奏控制消费端的核心是一个无限循环常见做法是while (true)加poll。poll的阻塞时长不能设成0否则会变成忙等CPU飙升也不能设得太大因为消费者在两次poll之间需要维持心跳虽然心跳由后台线程发送但长时间不返回poll会被判定为死亡。// ConsumerRunner.java 单线程消费循环 while (true) { ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(1000)); if (records.isEmpty()) { continue; } for (ConsumerRecordString, String record : records) { MessageEnvelope envelope objectMapper.readValue(record.value(), MessageEnvelope.class); if (dedupService.isDuplicate(envelope.getMsgId())) { log.info(重复消息跳过 msgId{}, envelope.getMsgId()); continue; } MessageHandler handler handlerRouter.get(envelope.getPayloadType()); handler.handle(envelope); dedupService.markProcessed(envelope.getMsgId()); } consumer.commitSync(); }这个循环的逻辑说明poll每次最多返回max.poll.records条处理完统一commitSync。dedupService是幂等层用msgId去重处理成功前不记录所以重复消费时最多重复处理一次。handlerRouter根据payloadType找到对应的处理器这是典型策略模式新增消息类型不需要改主循环。参数上要关注max.poll.interval.ms和业务处理时间的匹配。假如一条消息平均50毫秒100条要5秒远小于180秒安全但如果某条消息是耗时操作比如调外部接口超时20秒那100条累积起来可能超过180秒触发rebalance。解决思路有两个一是把max.poll.records调小到20变相缩短一轮处理时间二是把耗时操作放到独立线程池主线程用pause和resume控制分区暂停。第二种方案能保住吞吐但复杂度高建议先调小max.poll.records观察线上表现再决定要不要上线程池。3.3 顺序性与多线程消费partition级别的有序实现“Kafka消息顺序”这个点在Java面试题里反复出现答案其实固定Kafka只能保证单个分区内有序多分区之间不保证。所以设计消费端时要把“需要有序的消息”路由到同一个分区并且用同一个消费线程去处理。// 发送端同一业务ID的消息路由到同一分区 int partition Math.abs(businessId.hashCode()) % partitionCount; producer.send(new ProducerRecord(topic, partition, businessId, envelope.toJson()));这里用businessId.hashCode()取模算分区只要分区数不变同一个businessId就一定进同一个分区。但要注意hashCode取模在分区数扩容时会改变路由结果所以关键业务的排序消息要把分区数固定下来不要轻易扩容。消费端如果想多线程又保持顺序不能简单地开多线程并发处理全部消息应该按分区做分组每个分区一个线程或按分区取模分到多个线程保证同一个分区内的消息始终被同一个线程处理。// ConsumerThreadPool.java 按分区分配独立线程 MapInteger, ExecutorService partitionExecutors new ConcurrentHashMap(); for (ConsumerRecordString, String record : records) { ExecutorService executor partitionExecutors.computeIfAbsent( record.partition(), p - Executors.newSingleThreadExecutor()); executor.submit(() - handler.handle(record)); }这段代码的逻辑说明record.partition()是这条消息所在的分区号每个分区分配一个单线程执行器同一分区的消息串行处理。要注意partitionExecutors必须在rebalance时清理否则分区重新分配后旧线程还会处理该分区的历史消息造成数据被两个线程同时处理。另一个坑是线程池的队列长度如果某个分区消息堆积这个线程的队列会一直拖长其他分区不受影响但该分区延迟升高需要监控每个线程的队列深度。经验结论多线程消费能提升整体吞吐但顺序敏感的业务不能为了吞吐牺牲分区内串行。吞吐优先场景建议增加消费者实例数每个实例单线程消费顺序敏感场景建议保持分区与线程的固定映射。4. 重复消费、丢失与乱序Kafka源码设计避坑实录这一章写的是我见过最多的四类线上事故每一条都对应一套“现象、原因、解决”。如果只看Kafka官方文档你很难体会到这些参数在真实业务里是怎么互相拉扯的这里直接给结论。4.1 重复消费从现象到幂等设计的落地现象消费者处理了两次订单创建消息数据库里出现两条相同订单或者账户余额被扣了两次。Kafka的at-least-once语义下消费者崩溃、位点提交失败、rebalance都会导致同一批消息被重新消费生产端重试也会产生重复消息。解决不能靠业务判断必须用独立幂等标记。// DedupService 幂等实现Redis SETNX public boolean isDuplicate(String msgId) { return redis.setIfAbsent(kafka:dedup: msgId, 1, Duration.ofHours(24)); }用SETNX保证同一时刻只有一个消费者能处理这条msgId处理完成后不需要删除让key自然过期即可。过期时间要大于业务处理最长时间否则重试消息可能再次进入。更严格的方案是数据库唯一键把msgId作为业务表的唯一索引插入冲突就跳过。注意SETNX只能防住并发场景如果两个消费者在同一毫秒都拿到这条消息Redis能保证只有一个成功但业务处理完之后的markProcessed要放在事务提交之后否则会出现“Redis标记了但数据库没提交”的窗口。4.2 位移提交的边界手动提交时机与rebalance的碰撞现象某个消费组经常出现“消息处理成功但offset被重置到最早”导致积压几千条消息反复消费。原因之一是某次处理失败抛异常代码跳过了commitSync下一次poll时又因为auto.offset.resetearliest从头拉取。更常见的场景是rebalance发生时未提交的位点被回滚消费者对同一批消息重复处理。解决方式是把“处理成功”和“提交位点”绑紧先处理全部成功才提交提交放在finally里也要判断是否真的成功commitSync抛异常时不能吞掉。另外不要在commitAsync回调里做耗时操作异步提交的回调线程是Kafka客户端共享的阻塞它会影响心跳和下次拉取。我习惯在消费循环里用一个本地变量标记本轮是否全部成功成功才提交失败就break下一轮poll继续处理同一批。4.3 消息丢失acks与min.insync.replicas的配合现象生产环境broker发生磁盘故障主节点切换后发现最近几百条消息找不到。有人疑惑为什么配了acksall还是丢数据。原因acksall只保证消息被所有ISR副本写入但如果ISR里只有一个副本另外两个副本都挂了或同步滞后消息其实只写在主副本上主节点宕机照样丢。解决修改topic级别的min.insync.replicas强制要求至少2个副本同步才算写入成功。这个参数配合acksall才有真正的数据安全意义。如果生产端发送时发现可用副本不足会抛NotEnoughReplicasException此时要能感知到而不是盲目重试。创建一个可靠性优先的Topic时副本因子设3、min.insync.replicas设2既能容忍单节点故障又不牺牲太多可用性。# 创建可靠性优先的Topic kafka-topics.sh --create --topic order-event \ --partitions 12 --replication-factor 3 \ --config min.insync.replicas24.4 消息乱序重试与幂等生产者的边界现象同一订单的状态消息创建、支付、完成被消费者拿到的顺序是完成、支付、创建。原因生产端设置了retries某个分区请求超时后重试重试消息可能比下一条消息更晚到达broker同一分区内顺序被打乱。enable.idempotence解决的是重复不解决乱序。解决方式保持max.in.flight.requests.per.connection5且retries0时Kafka会保证同一分区内消息的重试顺序但如果业务手动指定了分区且发送线程并发对同一分区投递重试时仍可能乱序。订单状态这类强顺序消息我习惯在消息体里加sequence序号消费端维护一个“当前已处理的最大序号”小于等于最大序号的直接丢弃或告警。跨系统的最终一致场景序列号加版本号是最后一层防线比依赖Kafka的配置更可靠。5. 集群部署与参数调优让Kafka系统设计在Linux上站稳5.1 三节点集群的配置清单与部署顺序Kafka系统设计最终要跑在Linux服务器上部署方式决定后续维护成本。以3节点为例每个节点的config/server.properties至少要改broker.id、listeners、log.dirs、controller.quorum.voters。3.x以上版本建议直接用KRaft模式不再依赖ZooKeeper新项目能少维护一个中间件。# 节点1 KRaft模式关键配置 process.rolesbroker,controller node.id1 controller.quorum.voters1192.168.1.10:9093,2192.168.1.11:9093,3192.168.1.12:9093 listenersPLAINTEXT://192.168.1.10:9092 controller.listener.namesCONTROLLER advertised.listenersPLAINTEXT://192.168.1.10:9092 log.dirs/data/kafka/logs num.partitions6 default.replication.factor2 min.insync.replicas2 offsets.topic.replication.factor3参数说明num.partitions默认是3如果预期业务Topic很多建议默认升到6避免后来扩容分区带来的运维成本。default.replication.factor2是默认副本因子新建Topic不指定时会用这个值生产环境建议改3。offsets.topic.replication.factor3是消费者位移主题的副本数集群规模是3设成3最安全。advertised.listeners必须写客户端能访问到的地址很多部署踩坑都源于这里写了localhost。启动顺序KRaft模式先执行kafka-storage.sh format格式化存储目录再逐台启动。三台启动完成后运行kafka-topics.sh --describe确认每个Topic的ISR都正常。这里最常见的坑是多节点使用同一log.dirs目录或同node.id导致节点加入集群失败启动前要逐台核对。5.2 Java端必调参数吞吐与延迟的取舍表Java客户端的参数不是一个一个独立生效的它们会组合出不同行为。下面这张表是我常用的调整起点线上再按监控数据微调。参数默认值调整建议影响batch.size16KB64KB批次更大吞吐提升延迟不变linger.ms010~20ms攒批等待延迟增加吞吐提升acksallall数据安全优先不要为了性能改1max.poll.records500100~200调小降低单轮处理时间减少rebalance风险fetch.min.bytes11KB减少拉取请求次数降低网络开销compression.typenonelz4压缩率与CPU开销的平衡点max.partition.fetch.bytes1MB1MB单条消息超过1MB时再调大linger.ms和batch.size必须成对调只调大linger.ms但消息量少攒不够批次等于干等只调大batch.size但消息量多反而会把批次切得过大。这两个参数的目标是“在延迟容忍范围内尽量填满一个批次”。线上消息体如果平均2KB64KB批次大概需要32条消息才能填满按每秒几千条的生产速率20毫秒足够攒满。5.3 消息延迟高的排查路径从生产埋点到消费耗时Kafka消息延迟高很多人第一反应是查broker但大部分延迟其实出在Java设计层。排查顺序有一个固定套路第一步看MessageEnvelope里的timestamp用“消费时间减去生产时间”算出端到端延迟。第二步分段生产者发送完成时间和broker接收时间的差值看producer本地缓冲和网络broker写入时间和消费者拉取时间的差值看分区数与消费能力是否匹配消费者poll时间和业务处理完成时间的差值看消费端瓶颈。常见延迟原因有四个生产者linger.ms设得过大且消息量不够消息在本地攒批等超时消费者max.poll.records太大单轮处理时间长poll间隔被拉大Topic的num.partitions太少消费者组内实例数多于分区数多余实例空转业务线程池满了消费线程在等数据库连接poll被阻塞。提示延迟排查不要瞎猜先把MessageEnvelope的时间戳打全。生产消费两端时间差超过1秒时逐段打印耗时黑匣子立刻变成可观测链路。# 查看消费者组位点与延迟 kafka-consumer-groups.sh --bootstrap-server 192.168.1.10:9092 \ --group order-consumer --describe这条命令会列出每个分区的CURRENT-OFFSET、LOG-END-OFFSET和LAG。如果LAG持续增长说明消费速度跟不上生产速度优先看max.poll.records和消费线程数如果LAG为0但业务感知延迟高问题就在生产端或网络不在Kafka。6. 验证与监控给这套Java设计做一次端到端体检6.1 用Java写最小端到端验证用例接手或交付一套Kafka源码设计我习惯先写一个最小验证用例不启动业务依赖只验证“生产一条、消费一条、位点推进”。这个用例要能在开发环境直接跑排除数据库、Redis、第三方接口的干扰。// EndToEndVerify.java 最小端到端验证 Properties consumerProps new Properties(); consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, localhost:9092); consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, verify-group); consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, earliest); KafkaConsumerString, String consumer new KafkaConsumer(consumerProps); consumer.subscribe(Collections.singletonList(verify-topic)); producer.send(new ProducerRecord(verify-topic, key-1, hello-kafka)).get(); ConsumerRecordsString, String records consumer.poll(Duration.ofSeconds(5)); if (records.isEmpty()) { throw new IllegalStateException(端到端验证失败未消费到消息); } consumer.close();这个用例的逻辑说明发送端用get()阻塞等待发送结果消费端轮询等待只要在超时时间内拉到消息就算通过。它验证的是“客户端配置、网络、Topic创建”这一层基础设施是通的业务层代码不参与。再往后才是真正的集成测试发送1000条带msgId的消息消费端统计收到的总数和重复数断言重复数为0。6.2 压测经验值与硬件瓶颈Kafka读写上限在哪里Kafka的吞吐上限通常不在Kafka本身而在磁盘顺序写速度和网络带宽。单块SSD的顺序写能到几百MB每秒机械盘只有一百多MB12个分区的Topic在3节点集群上跑几十万条每秒的消息并不罕见但前提是生产端和消费端都有对应的并行度。分区数与读写性能的关系是分区越多并行度越高但分区超过某个值后文件句柄、内存占用、rebalance时间都会恶化。经验上单节点分区数建议控制在1000以内单Topic分区数与消费者实例数匹配消费者实例数不要超过分区数。压测时逐级加linger.ms和batch.size观察吞吐曲线找到平台期。监控方面除了kafka-consumer-groups.sh看LAG配合可视化工具看消费组位点和broker磁盘使用率能省掉大量人工登录排查。这套Java设计跑稳的关键是把生产端、消费端、位点都变成可观测的指标而不是出了问题再上服务器翻日志。这套方案我自己用下来的最大心得是Kafka最容易翻车的地方从来不是Kafka本身而是Java设计层对消费端状态的控制。手动提交位点、max.poll.records与处理时间的匹配、msgId幂等这三个点做扎实了线上能少接一半告警。希望帮到你。本文还有配套的精品资源点击获取 Kafka Java系统设计核心：生产消费封装、位点管理与避坑调优实战
简介这套基于Java语言的Kafka消息队列系统设计源码面向需要处理高吞吐量消息、构建实时数据管道或学习分布式中间件原理的Java开发者。项目共42个文件压缩包约77.3MB主要包含27个Java源文件、6个XML配置文件、1个YAML配置、属性文件、SQL脚本、Git忽略文件及Markdown文档等。Java文件覆盖生产者与消费者的核心逻辑、消息收发机制及Kafka集群交互XML与YAML配置文件用于设定服务器地址、数据库连接等关键参数SQL脚本可辅助初始化消息数据存储Markdown文档提供集群搭建指引方便快速理解架构并部署环境。当前已有316人学习下载。通过研读这套源码可以掌握Kafka在Java项目中的落地方式梳理负载均衡、故障转移、消息持久化等高级特性为自研消息队列或参与大数据平台开发积累直接可参考的代码与配置方案。1. Kafka的Java系统设计先搞懂它解决的三个问题再动手做Java后端绕不开消息队列选型时Kafka、RabbitMQ、RocketMQ各有拥趸但Kafka凭借高吞吐和分区模型几乎成了日志采集、流量削峰、数据同步场景的默认选项。所谓基于Java语言的Kafka消息队列系统设计源码本质是把生产端、消费端、位点、重试、幂等这些逻辑在Java工程里落地成一套可扩展的代码结构。它解决的核心问题只有三个消息怎么安全地发出去、消息怎么被正确地消费、系统出问题后怎么快速定位。适合正在做Kafka项目重构、或者打算把消息中间件从直连API升级为统一平台的团队。先说结论Kafka的运维坑远少于Java设计层的坑这套源码能否扛住生产关键在于你如何控制消费端的状态。2. Java侧的核心抽象设计从Topic到ConsumerGroup的物化Kafka本身是一个分布式的提交日志Topic是逻辑容器分区是物理并行单位ConsumerGroup是消费协调单位。Java工程落地时不能直接暴露原生客户端API给所有业务方原因有三个原生KafkaProducer和KafkaConsumer的生命周期管理容易失控每个业务方自行new实例连接数不可控各团队自行配置acks、序列化器、重试参数出问题时很难统一排查业务方直接操作ConsumerRecord会让消息体格式完全失控后续想加链路追踪、幂等标记都得改所有调用方。所以设计的第一步是在Java侧建立一层薄封装把Topic、消息体、ConsumerGroup这些概念物化成可复用的类。2.1 Producer与Consumer的接口设计为什么不该裸用原生API先看一个典型的反模式业务代码里直接new KafkaProducer发完消息不close下次调用再new一个。低并发时看不出问题一到高峰期文件描述符和内存直接被打满而且客户端没复用生产者的批量发送和缓冲机制完全失效。常见做法是封装一个MessageSender全局只持有一个KafkaProducer实例同时把序列化、发送失败补偿、监控上报都收口在这一层。// MessageSender.java 统一消息发送入口全局单例持有KafkaProducer Service public class MessageSender { private final KafkaProducerString, String producer; private final ObjectMapper mapper new ObjectMapper(); private final MessageRetryStore retryStore; // 本地失败补偿表 public MessageSender(KafkaProperties props) { Properties p new Properties(); p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, props.getBootstrapServers()); p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); p.put(ProducerConfig.ACKS_CONFIG, props.getAcks()); p.put(ProducerConfig.RETRIES_CONFIG, 3); p.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5); this.producer new KafkaProducer(p); } public void send(String topic, String key, Object body) { try { String value mapper.writeValueAsString(body); producer.send(new ProducerRecord(topic, key, value), (meta, ex) - { if (ex ! null) { log.error(kafka发送失败 topic{} key{}, topic, key, ex); retryStore.save(topic, key, value); // 入库等待定时重试 } }); } catch (JsonProcessingException e) { log.error(序列化失败 topic{}, topic, e); } } }这段代码的核心在于send方法是异步的KafkaProducer内部有缓冲区producer.send只负责把消息放入批次真正发往broker的是后台Sender线程。回调里的ex ! null代表发送失败此时把消息存入本地重试表由定时任务扫描重发而不是在回调里直接循环重试。参数说明里要注意MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION当同时设置retries 0时这个值必须大于1否则重试时会因为等待窗口为1而卡住吞吐设成5既能保留并发发送能力Kafka又保证同一个连接上最多5个未确认请求不会引发分区级乱序。2.2 消息模型与序列化设计一个可落地的消息体结构光有发送入口还不够得定消息体规范。Kafka的value可以是任意字节但生产系统里最怕的是“每个团队自己定义一套消息格式”。这套设计里我一般强制要求所有消息走同一个信封MessageEnvelope。// MessageEnvelope.java 通用消息信封 public class MessageEnvelope { private String msgId; // 全局唯一ID幂等用 private String sourceApp; // 来源系统 private long timestamp; // 生产时间 private String traceId; // 链路追踪ID private String payloadType; // 业务类型标识 private String payload; // 业务数据JSON private int retryCount; // 重试次数 }msgId用UUID还是雪花ID不重要关键是全局唯一它是消费端做幂等的唯一依据。traceId必须在入口生成并跨系统传递否则消息链路断了你根本不知道是哪一环延迟。payloadType字段存业务类型标识消费端根据它路由到对应的handleMessage方法。序列化层面payload固定用JSON字符串不直接在Kafka里传Java对象序列化后的二进制。原因有两个JSON对跨语言友好以后有Python或Go的消费者可以无缝接入Java原生序列化携带大量类元信息体积膨胀严重而且一旦类结构变更旧消息直接反序列化失败。如果后续单条消息超过几百KB、吞吐要求极高再考虑升级到Avro或Protobuf但初期用JSON能避免把设计复杂度带进消息链路。提示不要用ObjectMapper每次new一个实例它是线程安全的全局单例复用即可。序列化失败要打topic和payloadType不然线上排查不知道是哪类消息出的问题。2.3 ConsumerGroup与位点管理Java端的状态物化ConsumerGroup在Kafka里的含义是“一组协同消费订阅Topic的消费者”组内的分区分配策略决定了消息归谁。Java端设计源码时要理解Kafka保存的offset是分区的消费位点不是业务处理位点。// ConsumerConfig 手动提交位点 p.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); p.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, earliest); p.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100); p.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 180000);禁用自动提交后offset的提交完全由Java代码控制。常见做法是poll一批消息处理成功后才调用commitSync()如果处理失败不提交位点等待下次poll重新拉取。这里有一个关键参数MAX_POLL_INTERVAL_MS_CONFIG默认300秒。如果一批100条消息里有几条特别慢导致一轮处理超过300秒Kafka会认为该消费者已死亡主动触发rebalance把分区移交给其他消费者。这个设计本意是防止消费者假死但实际中经常成为“消息处理超时翻车”的元凶后面避坑章会专门讲。位点的另一个物化是重平衡监听器实现ConsumerRebalanceListener在分区被撤销或分配时记录日志、暂停或恢复业务处理确保在rebalance过程中不会出现“位点还没提交但分区已经移交”的窗口。// 注册rebalance监听器兜底提交位点 consumer.subscribe(Collections.singletonList(topic), new ConsumerRebalanceListener() { Override public void onPartitionsRevoked(CollectionTopicPartition partitions) { // 分区被收回前执行最后一次位点提交 consumer.commitSync(); } Override public void onPartitionsAssigned(CollectionTopicPartition partitions) { log.info(分区分配完成 partitions{}, partitions); } });这个监听器解决的是极端场景业务处理到一半被rebalance打断onPartitionsRevoked触发时主动提交一次把已处理但未提交的位点尽量推进缩小重复消费的窗口。注意commitSync在回调里执行会阻塞一小会儿但相比重复消费造成的业务影响这点阻塞是可接受的。3. 用Java封装Kafka的生产消费链路关键代码与参数说明上一章定的是抽象结构这一章落到可运行的生产消费链路。读者要有一个概念Kafka的Java客户端生产者是线程安全的一个KafkaProducer实例可以被多线程复用但KafkaConsumer不是线程安全的一个消费者实例只能由一个线程驱动。3.1 Producer端封装异步发送、回调与重试的配合生产者吞吐的瓶颈往往不在Kafka而在Java客户端的参数配合。下面是一组面向高吞吐的初始化配置。// ProducerInit.java 面向吞吐优先的生产者配置 Properties p new Properties(); p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 192.168.1.10:9092,192.168.1.11:9092,192.168.1.12:9092); p.put(ProducerConfig.ACKS_CONFIG, all); p.put(ProducerConfig.RETRIES_CONFIG, 3); p.put(ProducerConfig.BATCH_SIZE_CONFIG, 65536); // 64KB p.put(ProducerConfig.LINGER_MS_CONFIG, 20); p.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864); // 64MB p.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, lz4);BATCH_SIZE是单个分区的批次字节数上限默认16KB。如果单条消息只有几百字节这个值可以让多个消息凑成一个批次减少网络往返。LINGER_MS是批次等待时间默认0表示立即发送设成20毫秒意味着最多等20毫秒把更多消息攒进同一批吞吐会明显提升但消息延迟会增加20毫秒。BUFFER_MEMORY是生产者缓冲总内存默认32MB如果业务突发量很大这个值不够会导致send阻塞阻塞超过max.block.ms会抛TimeoutException。COMPRESSION_TYPE用lz4吞吐优先且CPU开销可控。这里要强调acksall配合retries3是安全性较高的组合但重试会带来重复消息所以消费端必须做幂等。enable.idempotence开启后生产者会自动处理重试产生的重复但有一个前提max.in.flight.requests.per.connection必须小于等于5且不能手动把acks配成其他值。如果业务要求消息不重不丢建议直接开幂等。重试的落地不是无限重试。我一般会在回调失败时把消息写入本地MySQL重试表用一个定时任务每30秒扫一次重试次数超过3次的置为失败状态由人工介入。这比在回调里sleep重试要稳得多因为回调线程是Kafka客户端内部的Sender线程阻塞它会影响所有消息发送。// RetryTask.java 定时补偿失败消息 Scheduled(fixedDelay 30000) public void retryFailedMessages() { ListMessageRetry retries retryStore.scan(3); for (MessageRetry r : retries) { try { producer.send(new ProducerRecord(r.getTopic(), r.getKey(), r.getValue())); retryStore.markSent(r.getId()); } catch (Exception e) { retryStore.incrRetryCount(r.getId()); } } }补偿任务的逻辑说明scan(3)只取重试次数小于3的消息发送成功就标记已处理发送失败就递增重试计数。这里要监控重试表的积压数量如果持续增长说明broker或Topic出了问题再增加重试次数只会放大故障不如直接告警。3.2 Consumer端拉取与处理poll循环的节奏控制消费端的核心是一个无限循环常见做法是while (true)加poll。poll的阻塞时长不能设成0否则会变成忙等CPU飙升也不能设得太大因为消费者在两次poll之间需要维持心跳虽然心跳由后台线程发送但长时间不返回poll会被判定为死亡。// ConsumerRunner.java 单线程消费循环 while (true) { ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(1000)); if (records.isEmpty()) { continue; } for (ConsumerRecordString, String record : records) { MessageEnvelope envelope objectMapper.readValue(record.value(), MessageEnvelope.class); if (dedupService.isDuplicate(envelope.getMsgId())) { log.info(重复消息跳过 msgId{}, envelope.getMsgId()); continue; } MessageHandler handler handlerRouter.get(envelope.getPayloadType()); handler.handle(envelope); dedupService.markProcessed(envelope.getMsgId()); } consumer.commitSync(); }这个循环的逻辑说明poll每次最多返回max.poll.records条处理完统一commitSync。dedupService是幂等层用msgId去重处理成功前不记录所以重复消费时最多重复处理一次。handlerRouter根据payloadType找到对应的处理器这是典型策略模式新增消息类型不需要改主循环。参数上要关注max.poll.interval.ms和业务处理时间的匹配。假如一条消息平均50毫秒100条要5秒远小于180秒安全但如果某条消息是耗时操作比如调外部接口超时20秒那100条累积起来可能超过180秒触发rebalance。解决思路有两个一是把max.poll.records调小到20变相缩短一轮处理时间二是把耗时操作放到独立线程池主线程用pause和resume控制分区暂停。第二种方案能保住吞吐但复杂度高建议先调小max.poll.records观察线上表现再决定要不要上线程池。3.3 顺序性与多线程消费partition级别的有序实现“Kafka消息顺序”这个点在Java面试题里反复出现答案其实固定Kafka只能保证单个分区内有序多分区之间不保证。所以设计消费端时要把“需要有序的消息”路由到同一个分区并且用同一个消费线程去处理。// 发送端同一业务ID的消息路由到同一分区 int partition Math.abs(businessId.hashCode()) % partitionCount; producer.send(new ProducerRecord(topic, partition, businessId, envelope.toJson()));这里用businessId.hashCode()取模算分区只要分区数不变同一个businessId就一定进同一个分区。但要注意hashCode取模在分区数扩容时会改变路由结果所以关键业务的排序消息要把分区数固定下来不要轻易扩容。消费端如果想多线程又保持顺序不能简单地开多线程并发处理全部消息应该按分区做分组每个分区一个线程或按分区取模分到多个线程保证同一个分区内的消息始终被同一个线程处理。// ConsumerThreadPool.java 按分区分配独立线程 MapInteger, ExecutorService partitionExecutors new ConcurrentHashMap(); for (ConsumerRecordString, String record : records) { ExecutorService executor partitionExecutors.computeIfAbsent( record.partition(), p - Executors.newSingleThreadExecutor()); executor.submit(() - handler.handle(record)); }这段代码的逻辑说明record.partition()是这条消息所在的分区号每个分区分配一个单线程执行器同一分区的消息串行处理。要注意partitionExecutors必须在rebalance时清理否则分区重新分配后旧线程还会处理该分区的历史消息造成数据被两个线程同时处理。另一个坑是线程池的队列长度如果某个分区消息堆积这个线程的队列会一直拖长其他分区不受影响但该分区延迟升高需要监控每个线程的队列深度。经验结论多线程消费能提升整体吞吐但顺序敏感的业务不能为了吞吐牺牲分区内串行。吞吐优先场景建议增加消费者实例数每个实例单线程消费顺序敏感场景建议保持分区与线程的固定映射。4. 重复消费、丢失与乱序Kafka源码设计避坑实录这一章写的是我见过最多的四类线上事故每一条都对应一套“现象、原因、解决”。如果只看Kafka官方文档你很难体会到这些参数在真实业务里是怎么互相拉扯的这里直接给结论。4.1 重复消费从现象到幂等设计的落地现象消费者处理了两次订单创建消息数据库里出现两条相同订单或者账户余额被扣了两次。Kafka的at-least-once语义下消费者崩溃、位点提交失败、rebalance都会导致同一批消息被重新消费生产端重试也会产生重复消息。解决不能靠业务判断必须用独立幂等标记。// DedupService 幂等实现Redis SETNX public boolean isDuplicate(String msgId) { return redis.setIfAbsent(kafka:dedup: msgId, 1, Duration.ofHours(24)); }用SETNX保证同一时刻只有一个消费者能处理这条msgId处理完成后不需要删除让key自然过期即可。过期时间要大于业务处理最长时间否则重试消息可能再次进入。更严格的方案是数据库唯一键把msgId作为业务表的唯一索引插入冲突就跳过。注意SETNX只能防住并发场景如果两个消费者在同一毫秒都拿到这条消息Redis能保证只有一个成功但业务处理完之后的markProcessed要放在事务提交之后否则会出现“Redis标记了但数据库没提交”的窗口。4.2 位移提交的边界手动提交时机与rebalance的碰撞现象某个消费组经常出现“消息处理成功但offset被重置到最早”导致积压几千条消息反复消费。原因之一是某次处理失败抛异常代码跳过了commitSync下一次poll时又因为auto.offset.resetearliest从头拉取。更常见的场景是rebalance发生时未提交的位点被回滚消费者对同一批消息重复处理。解决方式是把“处理成功”和“提交位点”绑紧先处理全部成功才提交提交放在finally里也要判断是否真的成功commitSync抛异常时不能吞掉。另外不要在commitAsync回调里做耗时操作异步提交的回调线程是Kafka客户端共享的阻塞它会影响心跳和下次拉取。我习惯在消费循环里用一个本地变量标记本轮是否全部成功成功才提交失败就break下一轮poll继续处理同一批。4.3 消息丢失acks与min.insync.replicas的配合现象生产环境broker发生磁盘故障主节点切换后发现最近几百条消息找不到。有人疑惑为什么配了acksall还是丢数据。原因acksall只保证消息被所有ISR副本写入但如果ISR里只有一个副本另外两个副本都挂了或同步滞后消息其实只写在主副本上主节点宕机照样丢。解决修改topic级别的min.insync.replicas强制要求至少2个副本同步才算写入成功。这个参数配合acksall才有真正的数据安全意义。如果生产端发送时发现可用副本不足会抛NotEnoughReplicasException此时要能感知到而不是盲目重试。创建一个可靠性优先的Topic时副本因子设3、min.insync.replicas设2既能容忍单节点故障又不牺牲太多可用性。# 创建可靠性优先的Topic kafka-topics.sh --create --topic order-event \ --partitions 12 --replication-factor 3 \ --config min.insync.replicas24.4 消息乱序重试与幂等生产者的边界现象同一订单的状态消息创建、支付、完成被消费者拿到的顺序是完成、支付、创建。原因生产端设置了retries某个分区请求超时后重试重试消息可能比下一条消息更晚到达broker同一分区内顺序被打乱。enable.idempotence解决的是重复不解决乱序。解决方式保持max.in.flight.requests.per.connection5且retries0时Kafka会保证同一分区内消息的重试顺序但如果业务手动指定了分区且发送线程并发对同一分区投递重试时仍可能乱序。订单状态这类强顺序消息我习惯在消息体里加sequence序号消费端维护一个“当前已处理的最大序号”小于等于最大序号的直接丢弃或告警。跨系统的最终一致场景序列号加版本号是最后一层防线比依赖Kafka的配置更可靠。5. 集群部署与参数调优让Kafka系统设计在Linux上站稳5.1 三节点集群的配置清单与部署顺序Kafka系统设计最终要跑在Linux服务器上部署方式决定后续维护成本。以3节点为例每个节点的config/server.properties至少要改broker.id、listeners、log.dirs、controller.quorum.voters。3.x以上版本建议直接用KRaft模式不再依赖ZooKeeper新项目能少维护一个中间件。# 节点1 KRaft模式关键配置 process.rolesbroker,controller node.id1 controller.quorum.voters1192.168.1.10:9093,2192.168.1.11:9093,3192.168.1.12:9093 listenersPLAINTEXT://192.168.1.10:9092 controller.listener.namesCONTROLLER advertised.listenersPLAINTEXT://192.168.1.10:9092 log.dirs/data/kafka/logs num.partitions6 default.replication.factor2 min.insync.replicas2 offsets.topic.replication.factor3参数说明num.partitions默认是3如果预期业务Topic很多建议默认升到6避免后来扩容分区带来的运维成本。default.replication.factor2是默认副本因子新建Topic不指定时会用这个值生产环境建议改3。offsets.topic.replication.factor3是消费者位移主题的副本数集群规模是3设成3最安全。advertised.listeners必须写客户端能访问到的地址很多部署踩坑都源于这里写了localhost。启动顺序KRaft模式先执行kafka-storage.sh format格式化存储目录再逐台启动。三台启动完成后运行kafka-topics.sh --describe确认每个Topic的ISR都正常。这里最常见的坑是多节点使用同一log.dirs目录或同node.id导致节点加入集群失败启动前要逐台核对。5.2 Java端必调参数吞吐与延迟的取舍表Java客户端的参数不是一个一个独立生效的它们会组合出不同行为。下面这张表是我常用的调整起点线上再按监控数据微调。参数默认值调整建议影响batch.size16KB64KB批次更大吞吐提升延迟不变linger.ms010~20ms攒批等待延迟增加吞吐提升acksallall数据安全优先不要为了性能改1max.poll.records500100~200调小降低单轮处理时间减少rebalance风险fetch.min.bytes11KB减少拉取请求次数降低网络开销compression.typenonelz4压缩率与CPU开销的平衡点max.partition.fetch.bytes1MB1MB单条消息超过1MB时再调大linger.ms和batch.size必须成对调只调大linger.ms但消息量少攒不够批次等于干等只调大batch.size但消息量多反而会把批次切得过大。这两个参数的目标是“在延迟容忍范围内尽量填满一个批次”。线上消息体如果平均2KB64KB批次大概需要32条消息才能填满按每秒几千条的生产速率20毫秒足够攒满。5.3 消息延迟高的排查路径从生产埋点到消费耗时Kafka消息延迟高很多人第一反应是查broker但大部分延迟其实出在Java设计层。排查顺序有一个固定套路第一步看MessageEnvelope里的timestamp用“消费时间减去生产时间”算出端到端延迟。第二步分段生产者发送完成时间和broker接收时间的差值看producer本地缓冲和网络broker写入时间和消费者拉取时间的差值看分区数与消费能力是否匹配消费者poll时间和业务处理完成时间的差值看消费端瓶颈。常见延迟原因有四个生产者linger.ms设得过大且消息量不够消息在本地攒批等超时消费者max.poll.records太大单轮处理时间长poll间隔被拉大Topic的num.partitions太少消费者组内实例数多于分区数多余实例空转业务线程池满了消费线程在等数据库连接poll被阻塞。提示延迟排查不要瞎猜先把MessageEnvelope的时间戳打全。生产消费两端时间差超过1秒时逐段打印耗时黑匣子立刻变成可观测链路。# 查看消费者组位点与延迟 kafka-consumer-groups.sh --bootstrap-server 192.168.1.10:9092 \ --group order-consumer --describe这条命令会列出每个分区的CURRENT-OFFSET、LOG-END-OFFSET和LAG。如果LAG持续增长说明消费速度跟不上生产速度优先看max.poll.records和消费线程数如果LAG为0但业务感知延迟高问题就在生产端或网络不在Kafka。6. 验证与监控给这套Java设计做一次端到端体检6.1 用Java写最小端到端验证用例接手或交付一套Kafka源码设计我习惯先写一个最小验证用例不启动业务依赖只验证“生产一条、消费一条、位点推进”。这个用例要能在开发环境直接跑排除数据库、Redis、第三方接口的干扰。// EndToEndVerify.java 最小端到端验证 Properties consumerProps new Properties(); consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, localhost:9092); consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, verify-group); consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, earliest); KafkaConsumerString, String consumer new KafkaConsumer(consumerProps); consumer.subscribe(Collections.singletonList(verify-topic)); producer.send(new ProducerRecord(verify-topic, key-1, hello-kafka)).get(); ConsumerRecordsString, String records consumer.poll(Duration.ofSeconds(5)); if (records.isEmpty()) { throw new IllegalStateException(端到端验证失败未消费到消息); } consumer.close();这个用例的逻辑说明发送端用get()阻塞等待发送结果消费端轮询等待只要在超时时间内拉到消息就算通过。它验证的是“客户端配置、网络、Topic创建”这一层基础设施是通的业务层代码不参与。再往后才是真正的集成测试发送1000条带msgId的消息消费端统计收到的总数和重复数断言重复数为0。6.2 压测经验值与硬件瓶颈Kafka读写上限在哪里Kafka的吞吐上限通常不在Kafka本身而在磁盘顺序写速度和网络带宽。单块SSD的顺序写能到几百MB每秒机械盘只有一百多MB12个分区的Topic在3节点集群上跑几十万条每秒的消息并不罕见但前提是生产端和消费端都有对应的并行度。分区数与读写性能的关系是分区越多并行度越高但分区超过某个值后文件句柄、内存占用、rebalance时间都会恶化。经验上单节点分区数建议控制在1000以内单Topic分区数与消费者实例数匹配消费者实例数不要超过分区数。压测时逐级加linger.ms和batch.size观察吞吐曲线找到平台期。监控方面除了kafka-consumer-groups.sh看LAG配合可视化工具看消费组位点和broker磁盘使用率能省掉大量人工登录排查。这套Java设计跑稳的关键是把生产端、消费端、位点都变成可观测的指标而不是出了问题再上服务器翻日志。这套方案我自己用下来的最大心得是Kafka最容易翻车的地方从来不是Kafka本身而是Java设计层对消费端状态的控制。手动提交位点、max.poll.records与处理时间的匹配、msgId幂等这三个点做扎实了线上能少接一半告警。希望帮到你。本文还有配套的精品资源点击获取