Kafka 是 Apache 基金会开发的分布式高吞吐消息队列，基于 Scala/Java 开发。核心设计包括分布式架构、磁盘顺序写、分区机制及消费者组模型。Java 开发主要使用 kafka-clients 原生客户端或 Spring Kafka 封装。典型应用涵盖微服务解耦、日志收集、流处理及削峰填谷。相比 RabbitMQ 和 RocketMQ，Kafka 在大数据和高吞吐场景更具优势，生产环境需注意序列化一致性、Offset 管理及集群配置。
Kafka 是Apache 基金会开发的一款分布式、高吞吐、低延迟、可持久化的分布式发布 - 订阅消息队列，基于 Scala/Java 开发，专为大数据场景下的高并发、高吞吐消息传输设计，是 Java 生态中最主流的分布式消息队列之一，广泛应用于日志收集、流处理、数据同步、微服务解耦等场景。
Kafka 集群由多个 Broker（服务节点） 组成，无中心节点，支持水平扩展（新增 Broker 即可提升集群能力）；通过副本机制（Replica）实现数据高可用，每个分区会有 1 个主副本（Leader）和多个从副本（Follower），Leader 负责读写，Follower 同步数据，Leader 故障时自动选举新 Leader，无数据丢失风险。
采用磁盘顺序写（磁盘顺序写性能接近内存）替代随机写，避免磁盘寻道开销；
引入页缓存（Page Cache） 机制，利用操作系统内存缓存消息，减少磁盘 I/O；
支持批量发送 / 拉取消息，减少网络请求次数，大幅提升吞吐能力（单节点可支撑十万级 TPS）；
低延迟特性，普通场景下消息从生产到消费的延迟可控制在毫秒级。
Kafka 会将所有消息持久化到磁盘（而非仅存内存），结合日志分段存储机制，消息可根据配置长期保留（数天 / 数月）；消费者不会删除消息，仅通过偏移量（Offset） 记录消费位置，支持任意时间点的回溯消费（比如重新消费几天前的日志数据）。
Kafka 的主题（Topic） 会被拆分为多个分区（Partition），分区是 Kafka 存储和并行处理的最小单位：
生产者（Producer）仅负责向 Topic 发送消息，无需关心消费者的数量、位置和消费状态；消费者（Consumer）仅从 Topic 拉取消息，无需关心生产者的情况，二者完全解耦，适合微服务架构中服务之间的异步通信。
理解以下核心概念是 Java 操作 Kafka 的基础，各概念关系为：集群包含多个 Broker → Broker 存储 Topic → Topic 拆分为多个 Partition → Partition 有多个 Replica → 消费者通过 Consumer Group 消费 Partition。
Kafka 为 Java 提供了官方原生客户端（kafka-clients），这是 Java 开发中最常用的客户端，轻量、高效，所有高级框架（如 Spring Kafka）都是基于它封装的。
<!-- Kafka Java 原生客户端核心依赖 -->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>3.6.1</version>
</dependency>
<!-- 若使用 Spring 生态，推荐使用 Spring Kafka（封装原生客户端，简化开发） -->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>3.1.1</version>
</dependency>
生产者：org.apache.kafka.clients.producer.KafkaProducer、ProducerRecord（封装待发送的消息，包含 Topic、分区、键值对）；
消费者：org.apache.kafka.clients.consumer.KafkaConsumer、ConsumerRecord（封装消费到的消息，包含 Topic、分区、Offset、键值对）；
配置类：生产者 / 消费者的配置通过 java.util.Properties 或 Spring 配置文件指定，核心配置有固定的常量（在 ProducerConfig、ConsumerConfig 中）。
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
public class KafkaProducerDemo {
// Kafka Broker 地址（集群用逗号分隔）
private static final String BOOTSTRAP_SERVERS = "127.0.0.1:9092";
// 要发送的 Topic
private static final String TOPIC = "test_topic";
public static void main(String[] args) {
// 1. 配置生产者参数
Properties props = new Properties();
// 必配：Broker 地址
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
// 必配：键的序列化器（消息的键和值必须序列化为字节数组，StringSerializer 序列化字符串）
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// 可选：消息确认机制（acks=1 表示主副本写入成功即返回，生产环境常用；acks=all 表示所有副本写入成功返回，最高可用）
props.put(ProducerConfig.ACKS_CONFIG, "1");
// 可选：发送失败重试次数
props.put(ProducerConfig.RETRIES_CONFIG, 3);
// 可选：批量发送大小（16KB），攒够大小再发送，提升吞吐
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
// 2. 创建生产者实例（实现 AutoCloseable，可使用 try-with-resources 自动关闭）
try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
// 3. 构建消息（ProducerRecord<键类型，值类型>）
ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, "key1", "hello kafka from java");
// 方式 1：异步发送（无返回，通过回调处理成功/失败）
producer.send(record, (metadata, exception) -> {
if (exception == null) {
// 发送成功：metadata 包含消息的分区、Offset 等信息
System.out.printf("异步发送成功：分区=%d，Offset=%d%n", metadata.partition(), metadata.offset());
} else {
// 发送失败：处理异常
exception.printStackTrace();
}
});
// 方式 2：同步发送（调用 get() 阻塞，直到返回结果）
try {
RecordMetadata metadata = producer.send(record).get();
System.out.printf("同步发送成功：分区=%d，Offset=%d%n", metadata.partition(), metadata.offset());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
// 刷新生产者，确保所有消息发送到 Broker
producer.flush();
}
}
}
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class KafkaConsumerDemo {
private static final String BOOTSTRAP_SERVERS = "127.0.0.1:9092";
private static final String TOPIC = "test_topic";
// 消费者组 ID（必须指定，相同组 ID 的消费者属于同一个消费者组）
private static final String GROUP_ID = "test_consumer_group";
public static void main(String[] args) {
// 1. 配置消费者参数
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
// 必配：键的反序列化器（与生产者序列化器对应）
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
// 可选：是否自动提交 Offset（默认 true，生产环境可改为 false，手动提交保证消费幂等）
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
// 可选：自动提交 Offset 的间隔（5000 毫秒）
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "5000");
// 可选：消费者启动时，若没有消费位置（首次消费），从最新消息开始消费（latest）/从最早消息开始消费（earliest）
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
// 2. 创建消费者实例
try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
// 3. 订阅 Topic（可订阅多个，传入集合）
consumer.subscribe(Collections.singletonList(TOPIC));
// 4. 循环拉取消息（消费者是拉取模式，需持续轮询）
while (true) {
// 拉取消息，超时时间 100 毫秒：若没有消息，等待 100 毫秒后返回空集合
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
// 遍历消费消息
for (ConsumerRecord<String, String> record : records) {
System.out.printf("消费成功：Topic=%s，分区=%d，Offset=%d，键=%s，值=%s%n",
record.topic(), record.partition(), record.offset(), record.key(), record.value());
}
}
}
}
}
order_topic，库存服务、支付服务、物流服务订阅该 Topic 分别处理，订单服务无需等待其他服务响应，提升系统吞吐量和容错性；Java 生态中常用的消息队列还有 RabbitMQ、RocketMQ，三者定位不同，Kafka 的核心优势体现在分布式、高吞吐、大数据场景，具体区别如下：
| 特性 | Kafka | RabbitMQ | RocketMQ |
|---|---|---|---|
| 核心定位 | 分布式高吞吐消息队列 | 轻量级 AMQP 协议队列 | 分布式通用消息队列 |
| 吞吐能力 | 极高（十万级 TPS） | 中等（万级 TPS） | 高（十万级 TPS） |
| 延迟特性 | 毫秒级 | 微秒级（低延迟） | 毫秒级 |
| 持久化 | 磁盘持久化，支持长期保留 | 磁盘 / 内存持久化 | 磁盘持久化 |
| 分布式架构 | 原生分布式，易扩展 | 需手动搭建集群，扩展较复杂 | 原生分布式，易扩展 |
| 协议支持 | 自定义协议 | AMQP/MQTT/STOMP | 自定义协议，兼容部分 MQTT |
| 适用场景 | 大数据、流处理、日志收集、削峰填谷 | 微服务轻量解耦、低延迟场景 | 微服务解耦、削峰填谷、分布式事务 |
| Java 集成 | 原生客户端 + Spring Kafka | 原生客户端 + Spring AMQP | 原生客户端 + Spring Cloud Stream |
总结：Kafka 不适合低延迟的轻量级微服务解耦（此场景优先 RabbitMQ），但在大数据、实时流处理、高吞吐日志收集等场景下，是 Java 生态的首选消息队列。
StringSerializer，消费者必须用 StringDeserializer）；复杂对象建议使用 Avro、Protobuf 序列化（而非 JSON），提升性能和兼容性；
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
查找任何按下的键的javascript键代码、代码、位置和修饰符。 在线工具，Keycode 信息在线工具，online
JavaScript 字符串转义/反转义；Java 风格 \uXXXX（Native2Ascii）编码与解码。 在线工具，Escape 与 Native 编解码在线工具，online
使用 Prettier 在浏览器内格式化 JavaScript 或 HTML 片段。 在线工具，JavaScript / HTML 格式化在线工具，online
Terser 压缩、变量名混淆，或 javascript-obfuscator 高强度混淆（体积会增大）。 在线工具，JavaScript 压缩与混淆在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online