Kafka基础理论与常用命令详解（超详细）

文章目录

前言
一、Kafka概述
- 1. Kafka简介
- 2. Kafka架构
- - 2.1 Broker（代理服务器）
  - 2.2 Topic（主题）
  - 2.3 Producer（生产者）
  - 2.4 Consumer（消费者）
  - 2.5 Consumer Group（消费者组）
- 3. Kafka特点
- - 3.1 高吞吐量
  - 3.2 持久性
  - 3.3 可扩展性
  - 3.4 实时性
  - 3.5 可靠性
- 4. Kafka应用场景
- - 4.1 实时数据流处理
  - 4.2 日志收集和聚合
  - 4.3 消息队列
  - 4.4 流式处理
  - 4.5 事件驱动架构
- 5. Kafka的副本机制
二、Topics命令介绍
- 1. Topics命令简介
- 2. Topics命令的基本语法
- 3. 常用的Topics命令选项
- 4. 常用的Topics命令示例
三、Producer命令介绍
- 1. Producer命令简介
- 2. Producer命令命令的基本语法：
- 3. 常用的Producer令选项
- 4. 常用的Producer命令示例
四、Consumer命令介绍
- 1. Consumer命令简介
- 2. Consumer命令的基本语法
- 3. 常用的Consumer命令选项
- 4. Consumer命令示例
五、Groups命令介绍
- 1. Groups命令简介
- 2. Groups命令的基本语法
- 3. 常用的Groups命令选项
- 4. 常用的Groups命令示例
六、Kafka安装教程
总结

前言

本文介绍了Kafka的基本概念和常用命令，包括Kafka的架构、特点和应用场景，以及Topics、Producer、Consumer和Groups命令的使用方法和常用选项。通过学习本文，您可以了解Kafka的基本原理和使用方法，以及如何使用命令行工具管理和操作Kafka集群。

一、Kafka概述

1. Kafka简介

Kafka是一个开源的分布式流处理平台，最初由LinkedIn开发并于2011年开源。它被设计用于高吞吐量、低延迟的数据传输，以及处理实时数据流。因其常被应用于消息队列，所以又被叫做分布式消息队列。

2. Kafka架构

2.1 Broker（代理服务器）

Kafka 集群中的每个节点都是一个独立的 Broker，它负责存储和处理消息。每个 Broker 可以在一个或多个Topic上发布和订阅消息。

2.2 Topic（主题）

消息在 Kafka 中以Topic的形式进行组织和分类。一个Topic可以有多个Partition，每个Partition在不同的 Broker 上进行复制，以实现高可用性和容错性。

Partition（分区）
每个Topic可以被分为多个Partition，每个Partition在磁盘上以一个日志文件进行存储。Partition允许消息在集群中进行并行处理，提高了吞吐量和扩展性。
offset（偏移量）
offset是用来唯一标识Partition中消息的位置信息。每个Partition都有自己的offset序列，用于标识消息在Partition中的顺序。
log segment（日志段）
Kafka使用一种称为log segment（日志段）的文件结构来保存消息。每个Topic的每个Partition都有一个或多个log segment，每个log segment都是一个连续的消息序列。当消息被写入Kafka时，它们会被追加到当前活动的log segment中。
index file（索引）
Kafka使用index file（索引文件）来加快消息的查找。index file（索引文件）包含了消息offset（偏移量）和物理文件位置之间的映射关系，以便能够快速定位特定offset（偏移量）的消息。

2.3 Producer（生产者）

Producer负责将消息发布到 Kafka 集群中的指定Topic。Producer可以选择将消息发送到特定的Partition，也可以使用Partitioner（分区器）根据某种策略自动选择分区。

2.4 Consumer（消费者）

Consumer可以订阅一个或多个Topic，并从Partition中拉取消息进行处理。消费者可以以不同的方式进行消息消费，例如，按照时间顺序、按照分区顺序或者以并行的方式消费消息。

2.5 Consumer Group（消费者组）

多个Consumer（消费者）可以组成一个Consumer Group（消费者组），共同消费一个Topic的消息。每个Partition只能由同一个Consumer Group（消费者组）中的一个Consumer（消费者）进行消费，这样可以实现负载均衡和水平扩展。

3. Kafka特点

3.1 高吞吐量

Kafka能够处理大规模的实时数据流，并具有非常高的吞吐量。它通过将数据分布在多个分区中，并允许并行写入和读取操作，实现了高度的并发性和可伸缩性。

3.2 持久性

Kafka将数据持久化到磁盘上，以确保数据的可靠性和持久性。它使用了顺序写入的方式来提高写入性能，并使用复制机制来保证数据的冗余备份，从而提供了高可靠性的数据存储。

3.3 可扩展性

Kafka的设计目标之一是能够轻松地扩展到大规模的集群。它通过分区和副本的概念来实现数据的分布和冗余备份，可以根据需求增加或减少分区和副本的数量，以适应不断增长的数据流量和负载。

3.4 实时性

Kafka是一个实时数据流平台，能够以毫秒级的延迟处理数据。它采用了发布-订阅模式，允许实时地将数据发布到主题（topics）中，并允许消费者实时订阅和处理这些数据。

3.5 可靠性

Kafka通过复制机制和分布式协调服务（如ZooKeeper）来提供高可靠性的数据存储和处理。它能够自动处理故障，包括节点故障、网络故障等，并保证数据的一致性和可用性。

4. Kafka应用场景

4.1 实时数据流处理

Kafka可以用作数据管道，将实时生成的数据流传输到不同的数据处理系统中。它可以接收大量的数据并将其持久化，同时允许多个消费者以实时方式订阅和处理这些数据。这种能力使得Kafka非常适合用于构建实时数据流处理和分析平台。

4.2 日志收集和聚合

Kafka可以用作日志收集和聚合的中间件。应用程序可以将日志消息发送到Kafka主题中，然后使用消费者来聚合、分析和存储这些日志数据。这种方式可以实现高可靠性的日志收集和处理，并支持实时监控和分析。

4.3 消息队列

Kafka作为消息队列系统，可以用于构建异步通信和解耦应用程序组件之间的通信。应用程序可以将消息发送到Kafka主题中，然后其他应用程序可以通过订阅这些主题来接收和处理消息。这种方式可以实现松耦合的系统架构，提高系统的可伸缩性和可靠性。

4.4 流式处理

Kafka可以与流处理框架（如Apache Flink、Apache Spark等）结合使用，构建实时流处理应用程序。Kafka作为数据源和数据接收器，可以提供可靠的数据传输和持久化，同时流处理框架可以通过Kafka的分区机制实现水平扩展和并行处理。

4.5 事件驱动架构

Kafka可以作为事件驱动架构的核心组件，用于实现事件的发布和订阅。应用程序可以将事件发送到Kafka主题中，然后其他应用程序可以通过订阅这些主题来接收和处理事件。这种方式可以实现松耦合、可扩展和可靠的事件驱动系统。

5. Kafka的副本机制

副本机制是指将消息主题的分区数据复制到多个Broker上，以确保数据的冗余和可靠性。在Kafka中，每个Partition可以有多个副本，其中一个副本被称为leader（领导者），其他副本被称为follower（追随者）。

以下是Kafka副本机制详解：

领导者和追随者角色
每个Partition都有一个leader和零个或多个follower。leader负责处理所有的读写请求，而follower只负责复制leader的数据。
数据复制
leader将消息写入本地日志，并将消息的副本发送给follower。follower接收到消息后，将其写入本地日志，并向leader发送确认消息。leader在收到足够数量的确认消息后，将消息标记为已提交。
ISR（In-Sync Replicas）
每个Partition的所有follower中，与leader保持同步的副本被称为ISR。只有ISR中的副本才能成为新的leader。如果一个follower与leader的同步延迟太大或无法与leader保持连接，它将被移出ISR。
副本选举
如果leader发生故障或无法正常工作，Kafka会自动进行副本选举，选择一个新的leader。副本选举过程中，只有ISR中的副本才有资格成为新的leader。
容错性
通过将消息的副本分布在多个Broker上，Kafka实现了容错性。即使某个Broker发生故障，其他副本仍然可以继续提供服务，确保数据的可用性和持久性。
扩展性
通过增加分区和副本的数量，Kafka可以实现水平扩展。更多的分区和副本可以提供更高的吞吐量和更好的负载均衡。

二、Topics命令介绍

1. Topics命令简介

Topics命令用于创建、列出和删除Kafka主题。您可以使用该命令来创建新的主题、查看现有主题的列表以及删除不再需要的主题。

2. Topics命令的基本语法

kafka-topics.sh --option1 value1 --option2 value2 ...

3. 常用的Topics命令选项

下面是Topics命令的常用选项：

--alter：修改主题的分区数、副本分配和/或配置。
--bootstrap-server <String: server to connect to>：必需选项，指定要连接的Kafka服务器。如果提供了此选项，则不需要直接连接到Zookeeper。
--command-config <String: command config property file>：指定包含要传递给Admin Client的配置的属性文件。此选项仅与--bootstrap-server选项一起使用，用于描述和修改代理配置。
--config <String: name=value>：为要创建或修改的主题提供配置覆盖。以下是一些有效的配置选项：
- cleanup.policy
- compression.type
- delete.retention.ms
- file.delete.delay.ms
- flush.messages
- flush.ms
- follower.replication.throttled.replicas
- index.interval.bytes
- leader.replication.throttled.replicas
- max.message.bytes
- message.downconversion.enable
- message.format.version
- message.timestamp.difference.max.ms
- message.timestamp.type
- min.cleanable.dirty.ratio
- min.compaction.lag.ms
- min.insync.replicas
- preallocate
- retention.bytes
- retention.ms
- segment.bytes
- segment.index.bytes
- segment.jitter.ms
- segment.ms
- unclean.leader.election.enable
请参阅Kafka文档以获取有关主题配置的完整详细信息。此选项仅在使用--bootstrap-server选项时与--create结合使用。
--create：创建新的主题。
--delete：删除主题。
--delete-config <String: name>：删除现有主题的配置覆盖。不支持与--bootstrap-server选项一起使用。
--describe：列出给定主题的详细信息。
--disable-rack-aware：禁用机架感知的副本分配。
--exclude-internal：在运行列表或描述命令时排除内部主题。默认情况下，内部主题将被列出。
--force：禁止控制台提示。
--help：打印使用信息。
--if-exists：如果设置了此选项并且要更改或删除或描述的主题存在，则仅执行操作。不支持与--bootstrap-server选项一起使用。
--if-not-exists：如果设置了此选项并且要创建的主题尚不存在，则仅执行操作。不支持与--bootstrap-server选项一起使用。
--list：列出所有可用的主题。
--partitions <Integer: # of partitions>：要创建或修改的主题的分区数（警告：如果增加具有键的主题的分区数，将影响分区逻辑或消息的顺序）。
--replica-assignment <String: broker_id_for_part1_replica1 : broker_id_for_part1_replica2 , broker_id_for_part2_replica1 : broker_id_for_part2_replica2 , ...>：为要创建或修改的主题提供手动分区到代理的分配列表。
--replication-factor <Integer: replication factor>：要创建的主题中每个分区的副本因子。
--topic <String: topic>：要创建、修改、描述或删除的主题。它还可以接受正则表达式，除了--create选项。将主题名称放在双引号中，并使用\前缀来转义正则表达式符号，例如：“test.topic”。
--topics-with-overrides：如果在描述主题时设置了此选项，则仅显示具有覆盖配置的主题。
--unavailable-partitions：如果在描述主题时设置了此选项，则仅显示其领导者不可用的分区。
--under-replicated-partitions：如果在描述主题时设置了此选项，则仅显示未充分复制的分区。
--zookeeper <String: hosts>：已弃用选项，指定Zookeeper连接字符串，格式为host:port。可以提供多个主机以实现故障转移。

4. 常用的Topics命令示例

以下是常用的Topics命令示例及其解释：

创建一个名为my_topic的主题：

kafka-topics.sh --create --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --partitions 3 --replication-factor 2

解释：使用--create选项创建一个名为my_topic的主题，该主题具有3个分区和2个副本。

--alter：修改名为my_topic的主题的分区数为5：

kafka-topics.sh --alter --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --partitions 5

解释：使用--alter选项修改名为my_topic的主题的分区数为5。

--bootstrap-server <String: server to connect to>：连接到Kafka服务器集群：
```
kafka-topics.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --list
```
解释：使用--bootstrap-server选项连接到Kafka集群，并列出所有可用的主题。
--command-config <String: command config property file>：使用名为admin.properties的属性文件连接到Kafka服务器：
```
kafka-topics.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --list --command-config admin.properties
```
解释：使用--command-config选项指定名为admin.properties的属性文件，该文件包含连接到Kafka服务器所需的配置信息，并列出所有可用的主题。
--describe：查看名为my_topic的主题的详细信息：
```
kafka-topics.sh --describe --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092"  --topic my_topic
```
解释：使用--describe选项查看名为my_topic的主题的详细信息，包括分区和副本信息。

--delete：删除名为my_topic的主题：

kafka-topics.sh --delete --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092"  --topic my_topic

解释：使用--delete选项删除名为my_topic的主题。
在这里插入图片描述

--exclude-internal：列出所有不包含内部主题的主题：
```
kafka-topics.sh --list --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092"   --exclude-internal
```
解释：使用--exclude-internal选项列出所有不包含内部主题的主题。
--force：删除名为my_topic的主题时禁止确认提示：
```
kafka-topics.sh --delete --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092"  --topic my_topic --force
```
解释：使用--force选项删除名为my_topic的主题时，禁止确认提示。
--topics-with-overrides：仅显示具有覆盖配置的主题的详细信息：
```
kafka-topics.sh --describe --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topics-with-overrides
```
解释：使用--topics-with-overrides选项仅显示具有覆盖配置的主题的详细信息。
--unavailable-partitions：仅显示其领导者不可用的分区的详细信息：
```
kafka-topics.sh --describe --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --unavailable-partitions
```
解释：使用--unavailable-partitions选项仅显示其领导者不可用的分区的详细信息。
--under-replicated-partitions：仅显示未充分复制的分区的详细信息：
```
kafka-topics.sh --describe --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --under-replicated-partitions
```
解释：使用--under-replicated-partitions选项仅显示未充分复制的分区的详细信息。
--zookeeper <String: hosts>：连接到Zookeeper服务器集群：
```
kafka-topics.sh --list --zookeeper "192.168.145.103:2181,192.168.145.104:2181,192.168.145.105:2181"
```
解释：使用--zookeeper选项连接到Zookeeper服务器集群，并列出所有可用的主题。
--help：打印帮助信息：
```
kafka-topics.sh --help
```
解释：使用--help选项打印关于Topics命令的帮助信息。

三、Producer命令介绍

1. Producer命令简介

kafka-console-producer.sh命令用于从命令行向Kafka主题发送消息。可以使用该命令将消息发送到指定的主题，以便进行测试和调试。

2. Producer命令命令的基本语法：

kafka-console-producer.sh --broker-list <broker-list> --topic <topic>

3. 常用的Producer令选项

--batch-size <Integer: size>：如果消息不是同步发送的，指定一次发送的消息批量大小。默认值为200。
--broker-list <String: broker-list>：必需选项，指定Kafka服务器的地址和端口列表，格式为HOST1:PORT1,HOST2:PORT2。
--compression-codec [String: compression-codec]：指定消息的压缩编解码器，可选值为’none’、‘gzip’、‘snappy’、‘lz4’或’zstd’。如果没有指定值，则默认为’gzip’。
--line-reader <String: reader_class>：指定用于从标准输入读取行的类名。默认情况下，每行被读取为一个单独的消息。
--max-block-ms <Long: max block on send>：生产者在发送请求期间阻塞的最长时间（以毫秒为单位）。默认值为60000。
--max-memory-bytes <Long: total memory in bytes>：生产者用于缓冲等待发送到服务器的记录的总内存大小。默认值为33554432字节（32MB）。
--max-partition-memory-bytes <Long: memory in bytes per partition>：为每个分区分配的缓冲区大小（以字节为单位）。当接收到小于此大小的记录时，生产者将尝试将它们乐观地组合在一起，直到达到此大小。默认值为16384字节（16KB）。
--message-send-max-retries <Integer>：代理服务器可能因多种原因而无法接收消息，而且临时不可用只是其中之一。此属性指定在生产者放弃并丢弃此消息之前的重试次数。默认值为3。
--metadata-expiry-ms <Long: metadata expiration interval>：在未看到任何领导者更改的情况下，强制刷新元数据的时间间隔（以毫秒为单位）。默认值为300000毫秒（5分钟）。
--producer-property <String: producer_prop>：以key=value形式传递自定义属性给生产者。
--producer.config <String: config file>：生产者配置属性文件。注意，[producer-property]优先于此配置。
--property <String: prop>：以key=value形式传递自定义属性给消息读取器。这允许为用户定义的消息读取器进行自定义配置。
--request-required-acks <String: request required acks>：生产者请求的必需确认级别。默认值为1。
--request-timeout-ms <Integer: request timeout ms>：生产者请求的确认超时时间。值必须为非负且非零值。默认值为1500毫秒。
--retry-backoff-ms <Integer>：在每次重试之前，生产者刷新相关主题的元数据。由于领导者选举需要一些时间，此属性指定生产者在刷新元数据之前等待的时间量。默认值为100毫秒。
--socket-buffer-size <Integer: size>：TCP接收缓冲区的大小。默认值为102400字节（100KB）。
--sync：如果设置了该选项，消息发送请求将以同步方式发送到代理服务器，一次只发送一个消息。
--timeout <Integer: timeout_ms>：如果设置了该选项，并且生产者运行在异步模式下，它指定了消息在等待足够的批量大小时排队的最大时间。值以毫秒为单位，默认为1000毫秒。
--topic <String: topic>：必需选项，指定要发送消息的目标主题。
--help：打印使用信息。

4. 常用的Producer命令示例

--broker-list <String: broker-list>：必需选项，指定Kafka服务器的地址和端口列表，格式为HOST1:PORT1,HOST2:PORT2。
```
kafka-console-producer.sh --broker-list "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092",192.168.145.104:9092,192.168.145.105:9092 --topic my_topic
```
解释：指定Kafka服务器的地址和端口列表，连接到包含三个Kafka服务器的集群。
--batch-size <Integer: size>：如果消息不是同步发送的，指定一次发送的消息批量大小。默认值为200。
```
kafka-console-producer.sh --broker-list "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --batch-size 100
```
解释：将消息批量大小设置为100，每次发送100条消息。
--compression-codec [String: compression-codec]：指定消息的压缩编解码器，可选值为’none’、‘gzip’、‘snappy’、‘lz4’或’zstd’。如果未指定值，则默认为’gzip’。
```
kafka-console-producer.sh --broker-list "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --compression-codec snappy
```
解释：将消息的压缩编解码器设置为’snappy’，以使用Snappy压缩算法进行消息压缩。
--max-block-ms <Long: max block on send>：生产者在发送请求期间阻塞的最大时间（以毫秒为单位）。默认值为60000。
```
kafka-console-producer.sh --broker-list "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --max-block-ms 5000
```
解释：将生产者在发送请求期间阻塞的最大时间设置为5000毫秒。
--max-memory-bytes <Long: total memory in bytes>：生产者用于缓冲等待发送到服务器的记录的总内存大小（以字节为单位）。默认值为33554432（32MB）。
```
kafka-console-producer.sh --broker-list "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --max-memory-bytes 67108864
```
解释：将生产者用于缓冲记录的总内存大小设置为67108864字节（64MB）。
--max-partition-memory-bytes <Long: memory in bytes per partition>：为每个分区分配的缓冲区大小（以字节为单位）。当接收到小于该大小的记录时，生产者会尝试将它们进行乐观地分组，直到达到该大小。默认值为16384（16KB）。
```
kafka-console-producer.sh --broker-list "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --max-partition-memory-bytes 32768
```
解释：将每个分区分配的缓冲区大小设置为32768字节（32KB）。
--message-send-max-retries <Integer>：在多种情况下，代理服务器可能因为多种原因无法接收消息，而临时不可用只是其中之一。该属性指定在生产者放弃并丢弃消息之前的重试次数。默认值为3。
```
kafka-console-producer.sh --broker-list "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --message-send-max-retries 5
```
解释：将消息发送的最大重试次数设置为5次。
--metadata-expiry-ms <Long: metadata expiration interval>：在没有看到任何领导者更改的情况下，强制刷新元数据的时间间隔（以毫秒为单位）。默认值为300000（5分钟）。
```
kafka-console-producer.sh --broker-list "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --metadata-expiry-ms 600000
```
解释：将元数据刷新的时间间隔设置为600000毫秒（10分钟）。
--producer-property <String: producer_prop>：以key=value形式传递用户定义的属性给生产者。
```
kafka-console-producer.sh --broker-list "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --producer-property acks=all
```
解释：将acks属性设置为all，以确保所有副本都确认接收消息。
--producer.config <String: config file>：生产者配置属性文件。注意，--producer-property选项优先于此配置。
```
kafka-console-producer.sh --broker-list "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --producer.config producer.properties
```
解释：使用producer.properties文件中的配置属性连接到Kafka服务器。
--property <String: prop>：以key=value形式传递用户定义的属性给消息读取器。这允许为用户定义的消息读取器进行自定义配置。
```
kafka-console-producer.sh --broker-list "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --property key=value
```
解释：将自定义的属性key=value传递给消息读取器。
--request-required-acks <String: request required acks>：生产者请求的所需确认级别。默认值为1。
```
kafka-console-producer.sh --broker-list "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --request-required-acks -1
```
解释：将请求的所需确认级别设置为-1，表示生产者等待所有副本都确认接收消息。
--request-timeout-ms <Integer: request timeout ms>：生产者请求的确认超时时间。值必须为非负且非零值。默认值为1500。
```
kafka-console-producer.sh --broker-list "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --request-timeout-ms 2000
```
解释：将请求的确认超时时间设置为2000毫秒。
--retry-backoff-ms <Integer>：在每次重试之前，生产者会刷新相关主题的元数据。由于领导者选举需要一些时间，该属性指定生产者在刷新元数据之前等待的时间量。默认值为100。
```
kafka-console-producer.sh --broker-list "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --retry-backoff-ms 200
```
解释：将生产者在刷新元数据之前等待的时间量设置为200毫秒。
--socket-buffer-size <Integer: size>：TCP接收缓冲区的大小。默认值为102400（100KB）。
```
kafka-console-producer.sh --broker-list "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --socket-buffer-size 204800
```
解释：将TCP接收缓冲区的大小设置为204800字节（200KB）。
--sync：如果设置了该选项，消息发送请求将以同步方式发送到代理服务器，一次只发送一个消息。
```
kafka-console-producer.sh --broker-list "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --sync
```
解释：设置消息发送请求为同步方式，一次只发送一个消息。
--timeout <Integer: timeout_ms>：如果设置了该选项，并且生产者运行在异步模式下，它指定了消息在等待足够的批量大小时排队的最大时间。值以毫秒为单位，默认为1000。
```
kafka-console-producer.sh --broker-list "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --timeout 500
```
解释：将消息排队的最大等待时间设置为500毫秒。
--topic <String: topic>：必需选项，指定要发送消息的目标主题。
```
kafka-console-producer.sh --broker-list "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic
```
解释：指定要发送消息的目标主题为my_topic。

四、Consumer命令介绍

1. Consumer命令简介

Consumer命令用于从Kafka主题中消费消息并在命令行中显示。通过该命令，您可以订阅指定的主题并实时查看消息的内容。

2. Consumer命令的基本语法

kafka-console-consumer.sh --bootstrap-server <server:port> --topic <topic> [options]

3. 常用的Consumer命令选项

--bootstrap-server <String: server to connect to>：REQUIRED: 要连接的Kafka服务器地址和端口。
--consumer-property <String: consumer_prop>：以键值对形式传递自定义的消费者属性。
--consumer.config <String: config file>：消费者配置属性文件。注意，[consumer-property]优先于此配置。
--enable-systest-events：记录消费者的生命周期事件，除了记录消费的消息之外。（这仅用于系统测试。）
--formatter <String: class>：用于格式化Kafka消息显示的类名。（默认值：kafka.tools.DefaultMessageFormatter）
--from-beginning：如果消费者没有已建立的偏移量，从日志中最早的消息开始消费，而不是最新的消息。
--group <String: consumer group id>：消费者所属的消费者组ID。
--help：打印帮助信息。
--isolation-level <String>：设置为"read_committed"以过滤未提交的事务消息。设置为"read_uncommitted"以读取所有消息。（默认值：read_uncommitted）
--key-deserializer <String: deserializer for key>：键的反序列化器。
--max-messages <Integer: num_messages>：消费的最大消息数量。如果未设置，将持续消费。
--offset <String: consume offset>：要消费的偏移量ID（非负数），或者"earliest"表示从开头开始，或者"latest"表示从末尾开始。（默认值：latest）
--partition <Integer: partition>：要消费的分区。如果未指定"–offset"，则从分区末尾开始消费。
--property <String: prop>：用于初始化消息格式化程序的属性。默认属性包括：
- print.timestamp=true|false：是否打印消息的时间戳。
- print.key=true|false：是否打印消息的键。
- print.value=true|false：是否打印消息的值。
- key.separator=<key.separator>：键的分隔符。
- line.separator=<line.separator>：行分隔符。
- key.deserializer=<key.deserializer>：键的反序列化器。
- value.deserializer=<value.deserializer>：值的反序列化器。
--skip-message-on-error：如果在处理消息时出现错误，跳过该消息而不是停止消费。
--timeout-ms <Integer: timeout_ms>：如果指定，当在指定的时间间隔内没有可消费的消息时，退出消费。
--topic <String: topic>：要消费的主题。
--value-deserializer <String: deserializer for values>：值的反序列化器。
--whitelist <String: whitelist>：使用正则表达式指定要包含在消费中的主题的白名单。

4. Consumer命令示例

--bootstrap-server <String: server to connect to>：指定连接到的Kafka服务器地址和端口。
```
kafka-console-consumer.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic
```
解释：使用--bootstrap-server选项连接到Kafka集群中的任意一个服务器，例如连接到地址为"192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092"的Kafka服务器。
--consumer-property <String: consumer_prop>：以键值对形式传递自定义的消费者属性。
```
kafka-console-consumer.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --consumer-property group.id=my_consumer_group
```
解释：使用--consumer-property选项传递自定义的消费者属性，例如设置消费者组ID为"my_consumer_group"。
--consumer.config <String: config file>：消费者配置属性文件。
```
kafka-console-consumer.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --consumer.config consumer.properties
```
解释：使用--consumer.config选项指定消费者配置属性文件，该文件包含消费者的配置信息。
--enable-systest-events：记录消费者的生命周期事件，除了记录消费的消息之外。
```
kafka-console-consumer.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --enable-systest-events
```
解释：使用--enable-systest-events选项记录消费者的生命周期事件，以及消费的消息。
--formatter <String: class>：用于格式化Kafka消息显示的类名。
```
kafka-console-consumer.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --formatter kafka.tools.DefaultMessageFormatter
```
解释：使用--formatter选项指定用于格式化Kafka消息显示的类名，例如使用kafka.tools.DefaultMessageFormatter进行格式化。
--from-beginning：从日志中最早的消息开始消费，而不是最新的消息。
```
kafka-console-consumer.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --from-beginning
```
解释：使用--from-beginning选项从指定主题的最早消息开始消费，而不是从最新消息开始。
--group <String: consumer group id>：消费者所属的消费者组ID。
```
kafka-console-consumer.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --group my_consumer_group
```
解释：使用--group选项指定消费者所属的消费者组ID，以便进行消费者组的管理和协调。
--isolation-level <String>：设置为"read_committed"以过滤未提交的事务消息。设置为"read_uncommitted"以读取所有消息。
```
kafka-console-consumer.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --isolation-level read_committed
```
解释：使用--isolation-level选项设置消费者的隔离级别，可以选择只消费已提交的事务消息或者消费所有消息。

--key-deserializer <String: deserializer for key>：键的反序列化器。

kafka-console-consumer.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --key-deserializer org.apache.kafka.common.serialization.StringDeserializer

解释：使用--key-deserializer选项指定键的反序列化器，以便正确解析和显示键的内容。

--max-messages <Integer: num_messages>：消费的最大消息数量。
```
kafka-console-consumer.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --max-messages 100
```
解释：使用--max-messages选项指定要消费的最大消息数量，消费达到指定数量后将停止消费。
--offset <String: consume offset>：要消费的偏移量ID（非负数），或者"earliest"表示从开头开始，或者"latest"表示从末尾开始。
```
kafka-console-consumer.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --offset earliest
```
解释：使用--offset选项指定要消费的偏移量，可以是具体的偏移量ID，或者使用"earliest"表示从开头开始，或者使用"latest"表示从末尾开始。
--partition <Integer: partition>：要消费的分区。
```
kafka-console-consumer.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --partition 0
```
解释：使用--partition选项指定要消费的分区，可以指定分区的编号进行消费。
--property <String: prop>：用于初始化消息格式化程序的属性。
```
kafka-console-consumer.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --property print.timestamp=true --property print.key=true
```
解释：使用--property选项初始化消息格式化程序的属性，例如设置打印消息的时间戳和键。
--skip-message-on-error：如果在处理消息时出现错误，跳过该消息而不是停止消费。
```
kafka-console-consumer.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --skip-message-on-error
```
解释：使用--skip-message-on-error选项在处理消息时出现错误时跳过该消息，继续消费下一条消息。
--timeout-ms <Integer: timeout_ms>：如果指定，当在指定的时间间隔内没有可消费的消息时，退出消费。
```
kafka-console-consumer.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --timeout-ms 5000
```
解释：使用--timeout-ms选项设置超时时间，如果在指定的时间间隔内没有可消费的消息，则退出消费。

--topic <String: topic>：要消费的主题。

kafka-console-consumer.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic

解释：使用--topic选项指定要消费的主题。

--value-deserializer <String: deserializer for values>：值的反序列化器。

kafka-console-consumer.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --topic my_topic --value-deserializer org.apache.kafka.common.serialization.StringDeserializer

解释：使用--value-deserializer选项指定值的反序列化器，以便正确解析和显示值的内容。

--whitelist <String: whitelist>：使用正则表达式指定要包含在消费中的主题的白名单。
```
kafka-console-consumer.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --whitelist "topic1|topic2"
```
解释：使用--whitelist选项使用正则表达式指定要包含在消费中的主题的白名单，例如匹配"topic1"和"topic2"的主题。
--help：打印帮助信息。
```
kafka-console-consumer.sh --help
```
解释：使用--help选项打印关于Consumer命令的帮助信息，包括所有可用选项和示例。

五、Groups命令介绍

1. Groups命令简介

kafka-consumer-groups.sh命令用于管理和查看Kafka消费者组。您可以使用该命令列出消费者组、查看消费者组的偏移量以及重置消费者组的偏移量等操作。

2. Groups命令的基本语法

kafka-consumer-groups.sh --bootstrap-server <server:port> [options]

3. 常用的Groups命令选项

--bootstrap-server <String: server to connect to>：指定连接到的Kafka服务器地址和端口。
--command-config <String: command config property file>：指定包含命令配置属性的属性文件。
--delete：删除消费者组的偏移量和所有权信息。
--describe：描述消费者组并列出与给定组相关的偏移量差距（尚未处理的消息数）。
--dry-run：仅显示结果，而不执行对消费者组的更改。
--execute：执行操作。
--export：将操作执行结果导出到CSV文件。
--from-file <String: path to CSV file>：从CSV文件中重置偏移量值。
--group <String: consumer group>：要操作的消费者组。
--help：打印帮助信息。
--list：列出所有消费者组。
--members：描述消费者组的成员信息。
--offsets：描述消费者组并列出组中所有主题分区及其偏移量差距。
--reset-offsets：重置消费者组的偏移量。
--state：描述消费者组的状态。
--timeout <Long: timeout (ms)>：设置某些用例的超时时间。
--to-current：将偏移量重置为当前偏移量。
--to-datetime <String: datetime>：将偏移量重置为指定日期时间之后的偏移量。
--to-earliest：将偏移量重置为最早的偏移量。
--to-latest：将偏移量重置为最新的偏移量。
--to-offset <Long: offset>：将偏移量重置为指定的偏移量。
--topic <String: topic>：要删除消费者组信息或包含在重置偏移量过程中的主题。
--verbose：提供附加信息，例如在描述组时显示更多详细信息。

4. 常用的Groups命令示例

--bootstrap-server <String: server to connect to>：指定连接到的Kafka服务器地址和端口。
```
kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --list
```
解释：使用--bootstrap-server选项连接到Kafka服务器，此示例连接到kafka集群的9092端口，并显示消费者信息。
--command-config <String: command config property file>：指定包含命令配置属性的属性文件。
```
kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --command-config consumer.properties --list
```
解释：使用--command-config选项指定包含命令配置属性的属性文件，此示例使用名为consumer.properties的属性文件。
--delete：删除消费者组的偏移量和所有权信息。
```
kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --delete --group my_consumer_group
```
解释：使用--delete选项删除名为my_consumer_group的消费者组的偏移量和所有权信息。
--describe：描述消费者组并列出与给定组相关的偏移量差距（尚未处理的消息数）。
```
kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --describe --group my_consumer_group
```
解释：使用--describe选项描述名为my_consumer_group的消费者组，并列出与该组相关的偏移量差距。
--dry-run：仅显示结果，而不执行对消费者组的更改。
```
kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --reset-offsets --group my_consumer_group --to-earliest --all-topics --dry-run
```
解释：使用--dry-run选项在重置名为my_consumer_group的消费者组的偏移量之前，仅显示计划的更改，而不实际执行更改。

--execute：执行操作。

kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --reset-offsets --group my_consumer_group --to-earliest --all-topics --execute

解释：使用--execute选项执行重置名为my_consumer_group的消费者组的偏移量的操作。

--export：将操作执行结果导出到CSV文件。

kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --reset-offsets --group my_consumer_group --to-earliest --all-topics --export --export-file reset_offsets.csv

解释：使用--export选项将重置名为my_consumer_group的消费者组的偏移量操作的结果导出到名为reset_offsets.csv的CSV文件。

--from-file <String: path to CSV file>：从CSV文件中重置偏移量值。
```
kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --reset-offsets --group my_consumer_group --from-file reset_offsets.csv --execute
```
解释：使用--from-file选项从名为reset_offsets.csv的CSV文件中读取偏移量值，并重置名为my_consumer_group的消费者组的偏移量。
--group <String: consumer group>：要操作的消费者组。
```
kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --describe --group my_consumer_group
```
解释：使用--group选项指定要操作的消费者组，此示例描述名为my_consumer_group的消费者组。
--help：打印帮助信息。
```
kafka-consumer-groups.sh --help
```
解释：使用--help选项打印关于kafka-consumer-groups.sh命令的帮助信息，包括所有可用选项和示例。
--list：列出所有消费者组。
```
kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --list
```
解释：使用--list选项列出连接到"192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092"的Kafka服务器上的所有消费者组。

--members：描述消费者组的成员信息。

kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --describe --group my_consumer_group --members

解释：使用--members选项显示名为my_consumer_group的消费者组的成员信息。

--offsets：描述消费者组并列出组中所有主题分区及其偏移量差距。
```
kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --describe --group my_consumer_group --offsets
```
解释：使用--offsets选项显示名为my_consumer_group的消费者组的偏移量信息。

--reset-offsets：重置消费者组的偏移量。

kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --reset-offsets --group my_consumer_group --to-earliest --all-topics --execute

解释：使用--reset-offsets选项重置名为my_consumer_group的消费者组的偏移量为最早的偏移量，并应用更改。

--state：描述消费者组的状态。

kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --describe --group my_consumer_group --state

解释：使用--state选项显示名为my_consumer_group的消费者组的状态信息。

--timeout <Long: timeout (ms)>：设置某些用例的超时时间。
```
kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --describe --group my_consumer_group --timeout 10000
```
解释：使用--timeout选项设置描述名为my_consumer_group的消费者组时的超时时间为10,000毫秒。

--to-current：将偏移量重置为当前偏移量。

kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --reset-offsets --group my_consumer_group --to-current --all-topics --execute

解释：使用--to-current选项将名为my_consumer_group的消费者组的偏移量重置为当前偏移量，并应用更改。

--to-datetime <String: datetime>：将偏移量重置为指定日期时间之后的偏移量。
```
kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --reset-offsets --group my_consumer_group --to-datetime "2023-01-01T00:00:00.000" --all-topics --execute
```
解释：使用--to-datetime选项将名为my_consumer_group的消费者组的偏移量重置为指定日期时间（2023年1月1日00:00:00.000之后）的偏移量，并应用更改。

--to-earliest：将偏移量重置为最早的偏移量。

kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --reset-offsets --group my_consumer_group --to-earliest --all-topics --execute

解释：使用--to-earliest选项将名为my_consumer_group的消费者组的偏移量重置为最早的偏移量，并应用更改。

--to-latest：将偏移量重置为最新的偏移量。

kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --reset-offsets --group my_consumer_group --to-latest --all-topics --execute

解释：使用--to-latest选项将名为my_consumer_group的消费者组的偏移量重置为最新的偏移量，并应用更改。

--to-offset <Long: offset>：将偏移量重置为指定的偏移量。
```
kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --reset-offsets --group my_consumer_group --to-offset 100 --topic my_topic --execute
```
解释：使用--to-offset选项将名为my_consumer_group的消费者组的偏移量重置为指定的偏移量（100），并应用更改。
--topic <String: topic>：要删除消费者组信息或包含在重置偏移量过程中的主题。
```
kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --delete --group my_consumer_group --topic my_topic
```
解释：使用--topic选项指定要删除消费者组信息或包含在重置偏移量过程中的主题（my_topic）。
--verbose：提供附加信息，例如在描述组时显示更多详细信息。
```
kafka-consumer-groups.sh --bootstrap-server "192.168.145.103:9092,192.168.145.104:9092,192.168.145.105:9092" --describe --group my_consumer_group --verbose
```
解释：使用--verbose选项在描述名为my_consumer_group的消费者组时提供更多详细信息。