大数据 -55 Kafka 实战详解：sh 启动、主题管理、Java 客户端完整示例

2025-07-30
山东
本文字数：4771 字
阅读完需：约 16 分钟

点一下关注吧！！！非常感谢！！持续更新！！！

🚀 AI 篇持续更新中！（长期更新）

AI 炼丹日志-30-新发布【1T 万亿】参数量大模型！Kimi‑K2 开源大模型解读与实践，持续打造实用 AI 工具指南！📐🤖

💻 Java 篇正式开启！（300 篇）

目前 2025 年 07 月 28 日更新到：Java-83 深入浅出 MySQL 连接、线程、查询缓存与优化器详解 MyBatis 已完结，Spring 已完结，Nginx 已完结，Tomcat 已完结，分布式服务正在更新！深入浅出助你打牢基础！

📊 大数据板块已完成多项干货更新（300 篇）：

包括 Hadoop、Hive、Kafka、Flink、ClickHouse、Elasticsearch 等二十余项核心组件，覆盖离线+实时数仓全栈！大数据-278 Spark MLib - 基础介绍机器学习算法梯度提升树 GBDT 案例详解

章节内容

上节我们完成了：

Kafka 介绍
ZK 的基本环境
Kafka 下载解压配置
Kafka 启动配置
Kafka 启动服务

Kafka 启动

上节我们通过 sh 脚本启动，但是当我们的 SSH 关闭的时候，Kafka 服务也退出。这里我们可以使用 Kakfa 的守护进程的方式启动，就可以在后台运行了。

kafka-server-start.sh -daemon /opt/servers/kafka_2.12-2.7.2/config/server.properties

复制代码

启动之后，我们可以通过 ps 工具看到：

ps aux | grep kafka

复制代码

返回结果如下图：

sh 脚本使用

topics.sh

kakfa-topics.sh 用于管理主题

查看所有

kafka-topics.sh --list --zookeeper h121.wzk.icu:2181

复制代码

当前执行返回的是空的，因为我们没有任何主题。

创建主题

kafka-topics.sh --zookeeper h121.wzk.icu:2181 --create --topic wzk_topic_1 --partitions 1 --replication-factor 1

复制代码

执行结果中，我们可以观察到，已经顺利的完成了。

查看主题

kafka-topics.sh --zookeeper h121.wzk.icu:2181 --describe --topic wzk_topic_1

复制代码

执行结果中，我们可以观察到，已经顺利的完成了。

删除主题

kafka-topics.sh --zookeeper h121.wzk.icu:2181 --delete --topic wzk_topic_1

复制代码

新建主题（用于测试）

kafka-topics.sh --zookeeper h121.wzk.icu:2181 --create --topic wzk_topic_test --partitions 1 --replication-factor 1

复制代码

producer.sh

kafka-console-producer.sh 用于生产消息

生成数据

kafka-console-producer.sh --topic wzk_topic_test --broker-list h121.wzk.icu:9092

复制代码

手动生成一批数据来进行测试：

consumer.sh

kafka-console-consumer.sh 用于消费消息

消费数据

kafka-console-consumer.sh --bootstrap-server h121.wzk.icu:9092 --topic wzk_topic_test

复制代码

此时，我们需要再开启一个 Producer 产生数据，它才会继续消费。

从头消费

kafka-console-consumer.sh --bootstrap-server h121.wzk.icu:9092 --topic wzk_topic_test --from-beginning

复制代码

从头开始消费的话，我们可以看到消费者已经把刚才我们写入的数据都消费了

Java API

架构图

POM

kafka-clients 是 Apache Kafka 提供的一个 Java 库，用于与 Kafka 进行交互。它是 Kafka 的核心组件之一，提供了对 Kafka 生产者和消费者的实现，使得 Java 应用程序可以方便地将数据发送到 Kafka 主题或从中读取数据。

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

复制代码

Producer（生产者）：
生产者是 Kafka 中负责发布消息的客户端组件。它通过 API 或 SDK 将消息发送到指定的 Kafka 主题（Topic）。每条消息由一个键值对（key-value pair）组成，其中键（key）用于确定消息的分区策略，而值（value）是实际的消息内容。Kafka 支持两种分区策略：
哈希分区（Hash-based Partitioning）：根据键的哈希值将消息分配到特定的分区，确保相同键的消息始终进入同一分区。
轮询分区（Round-robin Partitioning）：若未指定键，消息将均匀分配到各个分区。
示例：假设一个电商系统使用 Kafka 处理订单数据，生产者可以将订单 ID 作为键，确保同一订单的所有消息（如创建、支付、发货）都进入同一分区，保证顺序处理。
Consumer（消费者）：
消费者用于从 Kafka 的主题中订阅并消费消息。Kafka 支持**消费者组（Consumer Group）**机制，允许多个消费者协同工作以提高吞吐量。消费者组内的每个消费者负责读取特定分区的消息，实现负载均衡。
消费者组动态平衡：当消费者加入或离开组时，Kafka 会自动重新分配分区（Rebalance），确保每个分区仅由一个消费者处理。
消费偏移量（Offset）：消费者通过提交偏移量记录已消费的消息位置，支持从断点恢复。
应用场景：在日志分析系统中，多个消费者可以并行处理不同分区的日志数据，加速分析流程。
Topic（主题）：
主题是 Kafka 中消息的逻辑分类单元，类似于数据库中的表。生产者将消息发送到特定主题，消费者则订阅感兴趣的主题。
分区与扩展性：每个主题可配置多个分区（Partition），分区数决定了主题的并行处理能力。例如，一个包含 10 个分区的主题可支持最多 10 个消费者并发消费。
消息顺序性：消息在单个分区内严格有序，但不同分区间无顺序保证。若需全局有序，可将主题设为单分区（但会牺牲吞吐量）。
Partition（分区）：
分区是主题的物理分片，也是 Kafka 实现高吞吐和水平扩展的核心机制。
副本机制（Replication）：每个分区可配置多个副本（如 3 个），其中一个是 Leader，负责处理读写请求；其余为 Follower，同步 Leader 的数据。
Leader 选举：若 Leader 宕机，Kafka 会从 Follower 中选举新 Leader，确保高可用。
写入与存储：生产者将消息写入 Leader 副本后，Follower 通过 ISR（In-Sync Replicas）机制异步同步数据，确保数据冗余。
示例：在金融交易系统中，分区副本可防止单点故障导致数据丢失。

常用配置详解

1. bootstrap.servers

指定 Kafka broker 的地址列表，格式为 host1:port1,host2:port2,...。生产者和消费者通过这个地址连接到 Kafka 集群。这个配置支持多个 broker 地址，客户端会自动发现集群中的所有 broker。

示例：

bootstrap.servers=kafka1.example.com:9092,kafka2.example.com:9092,kafka3.example.com:9092

复制代码

2. 序列化相关配置

key.serializer / value.serializer

生产者需要将消息的键和值序列化为字节数组才能发送到 Kafka。常用序列化器包括：

org.apache.kafka.common.serialization.StringSerializer：字符串序列化
org.apache.kafka.common.serialization.ByteArraySerializer：字节数组序列化
org.apache.kafka.common.serialization.IntegerSerializer：整数序列化

示例配置：

key.serializer=org.apache.kafka.common.serialization.StringSerializervalue.serializer=org.apache.kafka.common.serialization.StringSerializer

复制代码

key.deserializer / value.deserializer

消费者需要将接收到的字节数组反序列化为可读的键和值。对应的反序列化器包括：

org.apache.kafka.common.serialization.StringDeserializer
org.apache.kafka.common.serialization.ByteArrayDeserializer
org.apache.kafka.common.serialization.IntegerDeserializer

3. acks（生产者配置）

控制生产者等待 broker 确认消息的级别，影响消息的持久性和吞吐量：

0：生产者不等待任何确认（最高吞吐量，最低可靠性）
1：等待 leader 确认（默认值，平衡可靠性和性能）
all或-1：等待所有 ISR 副本确认（最高可靠性，最低吞吐量）

应用场景：

日志收集等不严格要求消息不丢失的场景可用0
订单系统等关键业务建议使用all

4. enable.auto.commit（消费者配置）

控制消费者是否自动提交偏移量：

true（默认）：消费者后台自动定期提交
false：需要手动调用commitSync()或commitAsync()

建议：对于精确一次处理要求的场景，建议设为false并手动管理偏移量提交。

5. auto.offset.reset（消费者配置）

当消费者组没有初始偏移量或偏移量不存在时的处理策略：

earliest：从分区最早的消息开始消费
latest（默认）：从最新的消息开始消费
none：抛出异常

使用场景：

首次启动消费者时，earliest可以获取历史数据
实时处理场景通常使用latest
none适用于严格要求偏移量存在的场景

生产者 1 测试


public class TestProducer01 {
    public static void main(String[] args) throws Exception {        Map<String, Object> configs = new HashMap<>();        configs.put("bootstrap.servers", "h121.wzk.icu:9092");        configs.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");        configs.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");        configs.put("acks", "1");        KafkaProducer<Integer, String> producer = new KafkaProducer<>(configs);        ProducerRecord<Integer, String> record = new ProducerRecord<>(                "wzk_topic_test",                0, 0,                "hello world by java!"        );        Future<RecordMetadata> future = producer.send(record);        future.get(3_000, TimeUnit.SECONDS);        producer.close();    }
}

复制代码

生产者 1 运行

运行结果如下图：

消费者 01 运行


public class TestConsumer01 {
    public static void main(String[] args) throws Exception {        Map<String, Object> configs = new HashMap<>();        configs.put("bootstrap.servers", "h121.wzk.icu:9092");        configs.put("key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer");        configs.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");        configs.put("group.id", "wzk-test");
        KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(configs);
        final List<String> topics = Arrays.asList("wzk_topic_test");        consumer.subscribe(topics, new ConsumerRebalanceListener() {            @Override            public void onPartitionsRevoked(Collection<TopicPartition> collection) {                collection.forEach(item -> {                    System.out.println("剥夺的分区: " + item.partition());                });            }
            @Override            public void onPartitionsAssigned(Collection<TopicPartition> collection) {                collection.forEach(item -> {                    System.out.println("接收的分区: " + item.partition());                });            }        });
        final ConsumerRecords<Integer, String> records = consumer.poll(3_000);        final Iterable<ConsumerRecord<Integer, String>> topic1Iterable = records.records("topic_1");        topic1Iterable.forEach(record -> {            System.out.println("消息头字段：" + Arrays.toString(record.headers().toArray()));            System.out.println("消息的key：" + record.key());            System.out.println("消息的偏移量：" + record.offset());            System.out.println("消息的分区号：" + record.partition());            System.out.println("消息的序列化key字节数：" + record.serializedKeySize());            System.out.println("消息的序列化value字节数：" + record.serializedValueSize());            System.out.println("消息的时间戳：" + record.timestamp());            System.out.println("消息的时间戳类型：" + record.timestampType());            System.out.println("消息的主题：" + record.topic());            System.out.println("消息的值：" + record.value());        });
        consumer.close();    }
}

复制代码

消费者 01 测试

控制台运行截图如下：

发布于: 刚刚阅读数: 3

原文链接:【http://xie.infoq.cn/article/fda65731011a97afbe96e04a6】。文章转载请联系作者。

武子康

关注

永远好奇无限进步 2019-04-14 加入

Hi, I'm Zikang，好奇心驱动的探索者 | INTJ / INFJ 我热爱探索一切值得深究的事物。对技术、成长、效率、认知、人生有着持续的好奇心和行动力。坚信「飞轮效应」，相信每一次微小的积累，终将带来深远的改变。

发布

暂无评论

创作场景

大数据 -55 Kafka 实战详解：sh 启动、主题管理、Java 客户端完整示例

点一下关注吧！！！非常感谢！！持续更新！！！

🚀 AI 篇持续更新中！（长期更新）

💻 Java 篇正式开启！（300 篇）

📊 大数据板块已完成多项干货更新（300 篇）：

章节内容

Kafka 启动

sh 脚本使用

topics.sh

查看所有

创建主题

查看主题

删除主题

新建主题（用于测试）

producer.sh

生成数据

consumer.sh

消费数据

从头消费

Java API

架构图

POM

常用配置详解

1. bootstrap.servers

2. 序列化相关配置

key.serializer / value.serializer

key.deserializer / value.deserializer

3. acks（生产者配置）

4. enable.auto.commit（消费者配置）

5. auto.offset.reset（消费者配置）

生产者 1 测试

生产者 1 运行

消费者 01 运行

消费者 01 测试

武子康

评论