前言
本文介绍了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.policycompression.typedelete.retention.msfile.delete.delay.msflush.messagesflush.msfollower.replication.throttled.replicasindex.interval.bytesleader.replication.throttled.replicasmax.message.bytesmessage.downconversion.enablemessage.format.versionmessage.timestamp.difference.max.msmessage.timestamp.typemin.cleanable.dirty.ratiomin.compaction.lag.msmin.insync.replicaspreallocateretention.bytesretention.mssegment.bytessegment.index.bytessegment.jitter.mssegment.msunclean.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的消费者组时提供更多详细信息。
六、Kafka安装教程
Kafka安装教程：Kafka安装与配置-shell脚本一键安装配置（集群版）
总结
本文介绍了Kafka的基本概念和常用命令，包括Kafka的架构、特点和应用场景，以及Topics、Producer和Consumer命令的使用方法和常用选项。通过这些命令，可以方便地管理和操作Kafka集群，包括创建和删除主题、发送和消费消息，以及管理消费者组的偏移量等。
总的来说，Kafka是一个功能强大的分布式流处理平台，适用于处理大规模的实时数据流。通过合理使用Kafka的命令和功能，可以构建高吞吐量、低延迟的数据处理系统，实现实时数据流处理、日志收集和聚合、消息队列等应用场景。
希望本文对您有所帮助！如有任何疑问或问题，请随时在评论区留言。感谢阅读！