目录
- 前置内容
- 常用脚本说明和示例
- 描述与定义
- 部分术语说明
- 两种模式
- 点对点模式
- 发布订阅模式
- topic主题命令行操作指令
- 生产者
- Broker
- 消费者
前置内容
队列：
先进先出
应用：
大数据中主要用于离线和实时处理
流程：
Flume正常获取数据，但数据量不固定，时间不定期，如果直接写入到存储（HDFS）分析（Spark）等等可能遇到一定问题，因此先进入到Kafka集群。
其他同类型产品：
RabbitMQ、ActiveMQ、RocketMQ
总结描述：
kafka等消息队列实质是一个高性能的消息传输管道
常用脚本说明和示例
以下脚本均存放在kafka的bin目录下
| 脚本名称 | 作用 | 示例 |
|---|---|---|
connect-distributed.sh | 启动 Kafka Connect 的分布式模式，用于在 Kafka 集群之间复制数据或与其他系统集成。 | ./connect-distributed.sh connect-distributed.properties 启动使用指定配置文件的分布式模式。 |
connect-mirror-maker.sh | 启动 Kafka MirrorMaker，用于在两个 Kafka 集群之间镜像数据。 | ./connect-mirror-maker.sh mirror-maker.properties 启动使用指定配置文件的MirrorMaker。 |
connect-standalone.sh | 启动 Kafka Connect 的独立模式，适用于简单的数据集成任务。 | ./connect-standalone.sh connect-standalone.properties 启动使用指定配置文件的独立模式。 |
kafka-acls.sh | 管理 Kafka 的访问控制列表（ACLs），用于设置和查看主题、消费者组等资源的权限。 | ./kafka-acls.sh --list --topic test-topic 列出指定主题的ACLs。 |
kafka-broker-api-versions.sh | 查看 Kafka broker 支持的 API 版本。 | ./kafka-broker-api-versions.sh --bootstrap-server localhost:9092 查看本地broker的API版本。 |
kafka-configs.sh | 查看和修改 Kafka 主题、broker 或其他实体的配置。 | ./kafka-configs.sh --describe --topic test-topic 描述指定主题的配置。 |
kafka-console-consumer.sh | Kafka 控制台消费者，用于从 Kafka 主题中读取消息。 | ./kafka-console-consumer.sh --topic test-topic --bootstrap-server localhost:9092 消费消息。 |
kafka-console-producer.sh | Kafka 控制台生产者，用于向 Kafka 主题发送消息。 | ./kafka-console-producer.sh --topic test-topic --bootstrap-server localhost:9092 生产消息。 |
kafka-consumer-groups.sh | 查看和管理 Kafka 消费者组的信息，包括消费者组的偏移量、成员等。 | ./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group test-group 描述消费者组。 |
kafka-consumer-perf-test.sh | 运行 Kafka 消费者的性能测试，用于评估消费者的吞吐量。 | ./kafka-consumer-perf-test.sh --topic test-topic --bootstrap-server localhost:9092 --threads 1 --messages 1000 运行性能测试。 |
kafka-delete-records.sh | 删除 Kafka 主题中的特定消息记录，通常用于数据清理。 | ./kafka-delete-records.sh --bootstrap-server localhost:9092 --offset-json-file delete-offsets.json 根据偏移量文件删除记录。 |
kafka-delegation-tokens.sh | 管理 Kafka 的委托令牌，用于安全认证和授权。 | ./kafka-delegation-tokens.sh --bootstrap-server localhost:9092 --create --command-config token-command.properties 创建委托令牌。 |
kafka-dump-log.sh | 转储 Kafka 日志文件的内容，用于调试和分析。 | ./kafka-dump-log.sh --files /var/lib/kafka/data/test-topic-0/00000000000000000000.log 转储指定日志文件。 |
kafka-leader-election.sh | 触发 Kafka 分区的 leader 选举，用于故障恢复或负载均衡。 | ./kafka-leader-election.sh --bootstrap-server localhost:9092 --topic test-topic --partition 0 触发指定分区的leader选举。 |
kafka-log-dirs.sh | 查看 Kafka broker 的日志目录信息。 | ./kafka-log-dirs.sh --bootstrap-server localhost:9092 查看broker的日志目录信息。 |
kafka-mirror-maker.sh | （与 connect-mirror-maker.sh 类似）用于在两个 Kafka 集群之间镜像数据。 | 同connect-mirror-maker.sh示例。 |
kafka-preferred-replica-election.sh | 触发 Kafka 首选副本的选举，用于优化集群的副本分布。 | ./kafka-preferred-replica-election.sh --bootstrap-server localhost:9092 触发首选副本选举。 |
kafka-producer-perf-test.sh | 运行 Kafka 生产者的性能测试，用于评估生产者的吞吐量。 | ./kafka-producer-perf-test.sh --topic test-topic --num-records 1000000 --record-size 100 --throughput -1 --producer-props bootstrap.servers=localhost:9092 运行性能测试。 |
kafka-reassign-partitions.sh | 重新分配 Kafka 分区的副本，用于集群扩容、缩容或故障恢复。 | ./kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file partitions-to-move.json --execute 执行分区重分配。 |
kafka-replica-verification.sh | 验证 Kafka 分区副本的一致性，用于数据完整性检查。 | ./kafka-replica-verification.sh --bootstrap-server localhost:9092 --topic test-topic 验证指定主题的分区副本一致性。 |
kafka-run-class.sh | 运行任意的 Java 类，通常用于运行 Kafka 相关的 Java 工具或测试。 | ./kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic test-topic 运行GetOffsetShell工具。 |
kafka-server-start.sh | 启动 Kafka broker。 | ./kafka-server-start.sh server.properties 启动使用指定配置文件的broker。 |
kafka-server-stop.sh | 停止 Kafka broker。 | ./kafka-server-stop.sh 停止运行中的broker。 |
kafka-streams-application-reset.sh | 重置 Kafka Streams 应用程序的状态存储，用于调试或恢复。 | ./kafka-streams-application-reset.sh --application-id my-streams-app --bootstrap-servers localhost:9092 --zookeeper localhost:2181 重置指定应用程序的状态。 |
kafka-topics.sh | 管理 Kafka 主题，包括创建、删除、查看主题列表、修改主题配置等。 | ./kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1 创建主题。 |
kafka-verifiable-consumer.sh | 运行一个可验证的消费者，用于测试和验证 Kafka 消费者的行为。 | ./kafka-verifiable-consumer.sh --topic test-topic --bootstrap-server localhost:9092 --group-id test-group --messages 10 运行可验证的消费者。 |
kafka-verifiable-producer.sh | 运行一个可验证的生产者，用于测试和验证 Kafka 生产者的行为。 | ./kafka-verifiable-producer.sh --topic test-topic --bootstrap-server localhost:9092 --messages 10 运行可验证的生产者。 |
trogdor.sh | Kafka 的性能测试工具，用于评估 Kafka 集群的性能。 | ./trogdor.sh --broker-list localhost:9092 --test-props trogdor-test.properties 运行性能测试。 |
windows | （通常是一个目录）包含适用于 Windows 环境的 Kafka 脚本和工具。 | 无直接示例，但可以在Windows环境下找到并使用该目录下的脚本。 |
zookeeper-security-migration.sh | 用于迁移 Zookeeper 的安全配置，通常与 Kafka 的安全设置相关。 | ./zookeeper-security-migration.sh --zookeeper.server localhost:2181 --migration.properties zookeeper-migration.properties 执行安全配置迁移。 |
| `zookeeper-server-start |
描述与定义
部分术语说明
| 术语 | 描述 |
|---|---|
| 事件 | 记录世界或业务中“发生了某事”的事实，具有键、值、时间戳和可选元数据 |
| 事件键 | 事件的唯一标识符（如“白富美”） |
| 事件的值 | 事件的具体内容（如“向土豪支付了520元”） |
| 事件时间戳 | 事件发生的时间（如“yyyy年05月20日13:14”） |
| 生产者 | 向Kafka发布（写入）事件的客户端应用程序 |
| 消费者 | 订阅（读取和处理）Kafka中事件的客户端应用程序 |
| 主题 | 事件被组织并持久存储的地方，支持多生产者和多订阅者，一般只用于发布订阅模式 |
| - 主题特性 | 事件消费后不被删除，可根据配置保留一段时间 |
| Broker | Kafka服务器，负责持久化消息和传输消息 |
| - Broker功能 | 一台Kafka服务器即为一个broker，集群由多个broker组成 |
| 分区 | 主题被分成多个分区，分布在不同broker上，支持并发读写 |
| - 分区特性 | 具有相同事件键的事件被写入同一个分区，保证消费顺序与写入顺序一致 |
两种模式
点对点模式
一对一，消费者主动拉取数据，消息收到后消息清除
- 消息生产者生产消息发送到Queue中，然后消息消费者从Queue中取出并且消费消息。
- 消息被消费以后，queue中不再有存储，所以消息消费者不可能消费到已经被消费的消息。
- Queue支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费。
发布订阅模式
- 消息生产者(发布)将消息发布到topic中，可以有多个topic。
- 同时有多个消息消费者(订阅)消费该消息，消费数据后不删除。
- 每个消费者相互独立，都可以消费到数据，
topic主题命令行操作指令
| 参数 | 值 | 描述 |
|---|---|---|
| –bootstrap-server | node3:9092 | 连接的 Kafka Broker 主机名称和端口号 |
| –topic | <String: topic> 比如：topicA | 操作的 topic 名称 |
| –list | 查看所有主题 | |
| –create | 创建主题 | |
| –delete | 删除主题 | |
| –alter | 修改主题 | |
| –describe | 查看主题详细描述 | |
| –partitions | <Integer: # of partitions> | 设置分区数 |
| –replication-factor | <Integer: replication factor> | 设置分区副本 |
| –config | <String: name=value> | 更新系统默认的配置 |
| –version | 查看当前系统 Kafka 的版本 | |
| –help | 输出帮助信息 |
示例
- 创建
/opt/module/kafka/bin/kafka-topics.sh --bootstrap-server slave1:9092 --create --topic topicA --partitions 3 --replication-factor 2
- 查看所有主题
/opt/module/kafka/bin/kafka-topics.sh --bootstrap-server slave1:9092 --list
- 删除主题
/opt/module/kafka/bin/kafka-topics.sh --bootstrap-server slave1:9092 --delete --topic topicA
- 描述主题
/opt/module/kafka/bin/kafka-topics.sh --bootstrap-server slave1:9092 --describe --topic topicA
生产者
详见如下
生产者专题博客
Broker
Broker专题
消费者
消费者专题