Kafka Java案例怎么写？

wen python案例 2026-07-22 1

Kafka Java案例实战指南

目录导读

为什么选择Kafka？——消息队列的核心价值与适用场景
环境搭建与依赖配置——开发前的基石准备
生产者案例：从发送到确认——三种发送模式与回调机制
消费者案例：分区消费与偏移量管理——实现数据不丢失、不重复
高级特性实战——事务、幂等性与监控集成
常见问题与问答——面试级核心知识点澄清

为什么选择Kafka？

在微服务架构中，异步解耦与流量削峰是刚需，Kafka作为分布式消息中间件,具备以下不可替代的优势：

Kafka Java案例怎么写？

高吞吐：单机每秒可处理数十万条消息
持久化：消息写入磁盘，支持回溯消费
分区容错：通过副本机制保证数据可靠性

适用场景：用户行为日志采集、系统监控指标上报、订单状态变更通知、异步任务分发（如邮件/短信发送）。

环境搭建与依赖配置

1 启动Kafka服务

首先确保本地安装Kafka和ZooKeeper（Kafka 2.8+版本可配合KRaft模式脱离ZK）：

# 启动ZooKeeper（若使用）
bin/zookeeper-server-start.sh config/zookeeper.properties
# 启动Kafka
bin/kafka-server-start.sh config/server.properties

2 Maven依赖配置

在pom.xml中添加：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.6.1</version>
</dependency>

注意：避免同时引入多个版本,防止类冲突。

生产者案例：从发送到确认

核心逻辑：将数据序列化后发送到指定Topic。

1 基础生产者示例

import org.apache.kafka.clients.producer.*;
public class SimpleProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key1", "Hello Kafka");
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                System.out.println("发送成功，分区：" + metadata.partition() + "，偏移量：" + metadata.offset());
            } else {
                System.err.println("发送失败：" + exception.getMessage());
            }
        });
        producer.close();
    }
}

2 三种发送模式详解

模式	使用方式	可靠性	速度
拍发（fire-and-forget）	不调用回调	最低	最快
同步发送	`.get()`阻塞等待	中等	慢
异步回调	传入`Callback`	高（可重试）	推荐

最佳实践：生产环境务必使用异步回调并设置acks=all（全部副本确认）与retries=3。

消费者案例：分区消费与偏移量管理

关键点：消费者组保证每条消息只被组内一个实例消费。

1 消费者代码示例

import org.apache.kafka.clients.consumer.*;
public class SimpleConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 从头消费
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("my-topic"));
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                }
            }
        } finally {
            consumer.close();
        }
    }
}

2 偏移量提交策略

自动提交：enable.auto.commit=true，周期提交（可能重复消费）
手动同步提交：consumer.commitSync()，保证不丢失但性能低
手动异步提交：consumer.commitAsync()，配合重试机制最佳

错误案例：很多开发者忘记在消费失败时暂停偏移提交，导致数据丢失，解决方案：手动提交并捕获异常后调用consumer.pause()。

高级特性实战

1 事务性消息（Exactly-once）

producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("topic", "data"));
    producer.sendOffsetsToTransaction(offsets, groupId);
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

适用场景：支付系统、库存扣减等需要精确一次语义的业务。

2 幂等性生产者

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

原理：Kafka为每个生产者生成唯一ID,避免重复写入。

3 监控集成

通过JMX暴露Kafka指标，或用Prometheus + Grafana监控生产/消费速率：

# 开启JMX
KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=9999"

常见问题与问答

Q1：Kafka如何保证消息不丢失？
从生产端到消费端三层保障：

生产端：acks=all + retries>0
Broker端：replication.factor>=2，min.insync.replicas>=2
消费端：手动提交偏移量,处理完再提交。

Q2：分区数量如何设置？
按业务吞吐量计算：分区数 >= 最大消费线程数，同时建议大于等于Broker数以便均衡负载。
经验公式：预期TPS / 单个分区最大TPS = 必需分区数。

Q3：为什么我的消费者一直rebalance？
常见原因：

会话超时：session.timeout.ms设置过短（建议>15秒）
处理时间过长：调整max.poll.interval.ms或优化业务逻辑
心跳丢失：开启heartbeat.interval.ms小于session.timeout.ms的1/3

Q4：Kafka和RabbitMQ选型差异？

吞吐量：Kafka远高于RabbitMQ（Kafka: 100万+ TPS vs RabbitMQ: 2万+ TPS）
路由复杂度：RabbitMQ支持多路由键，Kafka通过分区实现简单路由
消息顺序：Kafka分区内有序，全局无序；RabbitMQ交换器级别可能乱序
运维成本：Kafka依赖ZK（或KRaft），组件更多

本文从安装配置到生产/消费者代码，再到事务、监控等高级特性，构建了一个完整的Kafka Java实战体系，核心要记住：生产端用异步回调+acks=all，消费端手动提交偏移量+合理处理重平衡，对于学习Kafka的开发者而言，多动手写代码并监控实际系统表现,远比记忆理论参数更有价值。

扩展阅读：Kafka Streams实时流处理、Schema Registry兼容性管理、Kafka Connect数据同步。