Kafka

Apache Kafka 是一个开源事件流平台，用于实时大规模收集、处理、存储和集成数据。它支持多种用例，包括流处理、数据集成和发布/订阅消息传递。

Kafka 最初由 LinkedIn 开发，于 2011 年开源，并于 2012 年成为 Apache 软件基金会项目。全球数千家组织使用它来支持关键任务实时应用程序，包括证券交易所、电子商务应用程序、物联网监控和分析等等。

相关书籍

有本叫《Kafka权威指南》 第2版，不过我也没看过，😄。

官方文档最好的教程

看什么书，不如看官方文档 https://kafka.apache.org/documentation ，把基础的东西掌握好，再看其他相关的书籍增强理论知识。学会基本概念后，一定要去看官方文档，增强自己。自上而下，先从整体简单内容入手，再细读文档。

入门

Getting Started

How Apache Kafka works.

简介

Introduction

什么是事件流

从技术角度来说，事件流处理是指以事件流的形式从数据库、传感器、移动设备、云服务和软件应用程序等事件源实时捕获数据；持久存储这些事件流以供日后检索；实时和回顾性地操作、处理和响应事件流；并根据需要将事件流路由到不同的目标技术。因此，事件流处理可确保数据的连续流动和解读，从而确保正确的信息在正确的时间出现在正确的位置。

Apache Kafka 是一个事件流平台

Kafka 结合了三种关键功能:

1. 发布（写入）和订阅（读取）事件流，包括从其他系统持续导入/导出数据。
2. 根据需要持久可靠地 存储 事件流。
3. 在事件流发生时或回顾性地 处理它们。

所有这些功能都以分布式、高度可扩展、弹性、容错且安全的方式提供。Kafka 可以部署在裸机硬件、虚拟机和容器上，也可以部署在本地和云端。您可以选择自行管理 Kafka 环境，也可以使用由多家供应商提供的完全托管服务。

Kafka是如何工作的

Kafka是一个分布式系统，由服务器和客户端组成，之间用TCP通信；

服务器：Kafka 由一台或多台服务器组成的集群运行，这些服务器可以跨越多个数据中心或云区域。其中一些服务器构成存储层，称为 代理 (broker)。其他服务器运行 Kafka Connect，以事件流的形式持续导入和导出数据，从而将 Kafka 与您现有的系统（例如关系数据库）以及其他 Kafka 集群集成。为了帮助您实现关键任务用例，Kafka 集群具有高度的可扩展性和容错能力：如果其中任何一台服务器发生故障，其他服务器将接管其工作，以确保持续运行且不会丢失任何数据。

客户端：允许您编写分布式应用程序和微服务，以便能够并行、大规模地读取、写入和处理事件流，即使在网络问题或机器故障的情况下也能保持容错能力。

主要概念和术语

向 Kafka 读取或写入数据时，您是以事件的形式进行的，如：

• 事件键：“爱丽丝”
• 事件值：“向鲍勃支付200美元”
• 事件时间戳：“2020年6月25日下午2:06”

生产者是向Kafka发布（写入）事件地客户端应用程序，而消费者是订阅（读取和处理）这些事件地应用程序。在 Kafka 中，生产者和消费者完全解耦，彼此互不影响，这是 Kafka 实现其高可扩展性的关键设计元素。例如，生产者无需等待消费者。Kafka 提供了各种保证，例如 “恰好一次” 处理事件的能力。

事件被组织并持久存储在主题（Topic）中，主题类似于文件系统中地文件夹，事件就是该文件夹中的文件。

Kafka 中的主题始终是多生产者和多订阅者的：一个主题可以有零个、一个或多个向其写入事件的生产者，以及零个、一个或多个订阅这些事件的消费者。

主题中的事件可以根据需要随时读取——与传统消息传递系统不同，事件在消费后不会被删除。

相反，您可以通过每个主题的配置设置来定义 Kafka 应保留事件的时间，超过此时间后，旧事件将被丢弃。Kafka 的性能相对于数据大小而言实际上是恒定的，因此长期存储数据完全没问题。

主题（Topic）是分区的，一个主题会分布在位于不同Kafka代理Broker上的多个“存储桶”中。

为了确保数据容错性和高可用性，每个主题都可以进行复制，甚至可以跨地理区域和数据中心进行复制。这样，在出现问题、需要对代理进行维护等情况下，始终有多个代理拥有数据副本。常见的生产环境设置是复制因子为 3，即始终有三个数据副本。

Kafka API

除了用于管理任务的命令行工具外，Kafka 还有五个核心 API：

• 管理 API：管理和检查主题、代理及其他 Kafka 对象。
• 生产者 API：将事件流发布（写入）到一个或多个 Kafka 主题。
• 消费者 API：订阅（读取）一个或多个主题，并处理其事件流。
• Kafka Streams API：实现流处理应用和微服务，提供高级函数（如转换、聚合、连接、有状态操作、窗口化、事件时间处理）。从输入主题读取数据，生成输出主题，实现流转换。
• Kafka Connect API：构建和运行数据导入/导出连接器，实现 Kafka 与外部系统（如 PostgreSQL）的集成（读取/写入事件流）。社区已提供数百个现成连接器，通常无需自定义。

用例

Use Cases

1. 消息传递

Kafka 能有效替代传统消息代理（如 ActiveMQ、RabbitMQ），其高吞吐量、内置分区、复制和容错功能远超大多数系统。消息代理常用于解耦生产者与处理过程、缓冲消息等，是大规模消息处理应用的理想方案。

2. 网站追踪

Kafka 最初用于重建用户活动跟踪管道为实时发布-订阅数据流：网站活动（如页面浏览、搜索）发布到对应主题，可订阅用于实时处理、监控，或加载到 Hadoop/数据仓库进行离线分析。活动数据量巨大，每页浏览生成多条消息。

3. 指标

Kafka 常用于操作监控数据。这涉及聚合来自分布式应用程序的统计数据，以生成集中式的操作数据。

4. 日志聚合

Kafka 常替代日志聚合方案，后者从服务器收集物理日志文件并集中存储（如 HDFS）处理。Kafka 将日志/事件抽象为消息流，降低延迟，支持多数据源和分布式消费。相较 Scribe/Flume 等系统，Kafka 性能相当，但提供更强复制持久性和更低端到端延迟。

5. 流处理

许多 Kafka 用户通过多阶段处理管道处理数据：原始输入从 Kafka 主题消费，经聚合、丰富或转换后发布到新主题，用于进一步消费。例如，新闻推荐管道：RSS 抓取文章到“文章”主题；规范化/去重后发布到新主题；最终推荐给用户，形成实时数据流图。从 0.10.0.0 版本起，Kafka 提供轻量级流处理库 Kafka Streams 执行此类操作；其他开源工具包括 Apache Storm 和 Apache Samza。

6. 事件溯源

事件溯源是一种应用程序设计风格，其中状态变化以按时间顺序排列的记录序列进行记录。Kafka 支持存储海量日志数据，使其成为此类应用程序的理想后端。

7. 提交日志

Kafka 可作为分布式系统的外部提交日志，用于节点间数据复制和故障节点恢复的重新同步；其日志压缩功能支持此用法，类似于 Apache BookKeeper。

快速入门

Quick Start

哪些手动部署的搭建环境的方式我们直接跳过，我们直接用 docker-compose

Docker Compose搭建Kafka集群环境

拉取Docker镜像

root@ser745692301841:/dev_dir/note# docker pull apache/kafka:4.1.0
root@ser745692301841:/dev_dir/note# docker images
REPOSITORY     TAG          IMAGE ID       CREATED        SIZE
apache/kafka   4.1.0        a183a690a3a6   7 weeks ago    437MB

https://hub.docker.com/r/apache/kafka 仓库中的概述中 对于怎么使用写得很清楚。直接部署多节点的集群。

该部署由3个Broker和3个Controller组成，它们分别在各自的容器中运行(即KRaft隔离模式)。在Docker中进行此练习可以方便地了解多Broker配置和Kafka协议， 但此Docker Compose示例不适用于生产部署。

kafka/docker-compose.yml

services:
  controller-1:
    image: apache/kafka:4.1.0
    container_name: controller-1
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: controller
      KAFKA_LISTENERS: CONTROLLER://:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:9093,2@controller-2:9093,3@controller-3:9093
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0

  controller-2:
    image: apache/kafka:4.1.0
    container_name: controller-2
    environment:
      KAFKA_NODE_ID: 2
      KAFKA_PROCESS_ROLES: controller
      KAFKA_LISTENERS: CONTROLLER://:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:9093,2@controller-2:9093,3@controller-3:9093
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0

  controller-3:
    image: apache/kafka:4.1.0
    container_name: controller-3
    environment:
      KAFKA_NODE_ID: 3
      KAFKA_PROCESS_ROLES: controller
      KAFKA_LISTENERS: CONTROLLER://:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:9093,2@controller-2:9093,3@controller-3:9093
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0

  broker-1:
    image: apache/kafka:4.1.0
    container_name: broker-1
    ports:
      - 29092:9092
    environment:
      KAFKA_NODE_ID: 4
      KAFKA_PROCESS_ROLES: broker
      KAFKA_LISTENERS: 'PLAINTEXT://:19092,PLAINTEXT_HOST://:9092'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://broker-1:19092,PLAINTEXT_HOST://127.0.0.1:29092'
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:9093,2@controller-2:9093,3@controller-3:9093
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
    depends_on:
      - controller-1
      - controller-2
      - controller-3

  broker-2:
    image: apache/kafka:4.1.0
    container_name: broker-2
    ports:
      - 39092:9092
    environment:
      KAFKA_NODE_ID: 5
      KAFKA_PROCESS_ROLES: broker
      KAFKA_LISTENERS: 'PLAINTEXT://:19092,PLAINTEXT_HOST://:9092'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://broker-2:19092,PLAINTEXT_HOST://127.0.0.1:39092'
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:9093,2@controller-2:9093,3@controller-3:9093
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
    depends_on:
      - controller-1
      - controller-2
      - controller-3

  broker-3:
    image: apache/kafka:4.1.0
    container_name: broker-3
    ports:
      - 49092:9092
    environment:
      KAFKA_NODE_ID: 6
      KAFKA_PROCESS_ROLES: broker
      KAFKA_LISTENERS: 'PLAINTEXT://:19092,PLAINTEXT_HOST://:9092'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://broker-3:19092,PLAINTEXT_HOST://127.0.0.1:49092'
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:9093,2@controller-2:9093,3@controller-3:9093
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
    depends_on:
      - controller-1
      - controller-2
      - controller-3

Kafka3.0+版本引入地角色分离模式，将控制器和代理分离运行，控制器专注于控制平面，broker 专注于数据平面。整个集群总共 6 个节点（Node ID 从 1 到 6）：

• 3个专用控制器节点：负责元数据管理、Leader选举和集群协调 使用Raft协议实现quorum
• 3个专用代理节点：负责实际地消息存储、读写、复制

控制器节点(controller-1、controller-2、controller-3)

这三个服务是专用控制器，每个运行一个Kafka进程，但仅扮演controller角色，它们形成一个3节点quorum，确保高可用（至少2个节点存活即可维持集群）

• 共同配置：
  • image: apache/kafka:4.1.0：使用官方 Kafka Docker 镜像。
  • container_name：容器命名，便于管理（如 controller-1）。
  • KAFKA_PROCESS_ROLES: controller：指定进程仅作为控制器运行，不处理 broker 职责。
  • KAFKA_LISTENERS: CONTROLLER://:9093：监听控制器端口 9093，用于内部 quorum 通信。
  • KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT：内部 broker 通信使用 PLAINTEXT 协议（无加密）。
  • KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER：定义控制器监听器的名称。
  • KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:9093,2@controller-2:9093,3@controller-3:9093：定义 quorum 投票成员列表。格式为 <node_id>@<hostname>:<port>，所有控制器互相投票选举元数据领导者。
  • KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0：消费者组再平衡初始延迟设为 0ms，加速测试环境启动（生产环境可调整）。
• 差异：
  • 每个控制器的 KAFKA_NODE_ID 不同（1、2、3），确保节点唯一标识。
• 作用：控制器管理集群元数据（如 topic 分区分配、领导者选举），不存储消息。启动时，它们会通过 9093 端口形成 Raft 共识群。

代理节点(broker-1、broker-2、broker-3)

这三个服务是专用代理，每个运行一个Kafka进程，扮演broker角色，它们连接到控制器 quorum，处理消息生产/消费。

• 共同配置：
  • image: apache/kafka:4.1.0：同上。
  • container_name：容器命名（如 broker-1）。
  • ports: - <host_port>:9092：将容器内 9092 端口映射到主机（29092、39092、49092），允许外部客户端（如 Kafka 工具）连接。注意：主机端口不同，避免冲突。
  • KAFKA_PROCESS_ROLES: broker：指定进程仅作为 broker 运行，不处理控制器职责。
  • KAFKA_NODE_ID：唯一 ID（4、5、6）。
  • KAFKA_LISTENERS: 'PLAINTEXT://:19092,PLAINTEXT_HOST://:9092'：定义两个监听器：PLAINTEXT://:19092：内部监听器，用于 broker 间通信（容器网络内）。PLAINTEXT_HOST://:9092：外部监听器，用于主机连接。
  • KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://broker-<n>:19092,PLAINTEXT_HOST://localhost:<host_port>'：广告监听器，告诉客户端如何连接。内部用容器名（broker-1 等），外部用 localhost:<host_port>。
  • KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT：broker 间通信使用内部 PLAINTEXT 监听器（19092）。
  • KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER：broker 连接控制器的监听器名称（尽管 broker 不运行控制器，但需声明以支持未来扩展）。
  • KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT：映射监听器名称到协议（全 PLAINTEXT，无安全）。
  • KAFKA_CONTROLLER_QUORUM_VOTERS：同控制器，broker 需知道 quorum 地址以注册集群。
  • KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0：同上。
  • depends_on: - controller-1 - controller-2 - controller-3：确保 broker 在所有控制器启动后才启动，避免连接失败。
• 差异：
  • 每个 broker 的 ports、KAFKA_ADVERTISED_LISTENERS 中的主机端口不同（29092、39092、49092），以及容器名和 Node ID。
• 作用：
  • broker 处理实际数据流量。客户端通过主机端口连接（如 localhost:29092 连 broker-1），broker 内部通过 19092 端口复制数据。默认无 topic 配置，需手动创建。

整体架构与注意事项

• 网络通信：
  • 内部：所有容器在同一 Docker 网络中，使用服务名（如 broker-1）解析地址。控制器用 9093 端口 quorum，broker 间用 19092。
  • 外部：仅 broker 暴露端口到主机，控制器不暴露（纯内部）。
  • 协议全为 PLAINTEXT（无 TLS/认证），适合开发/测试；生产需添加 SASL/SSL。
• 启动流程：
  • 运行 docker-compose up -d：先启动 3 个控制器，形成 quorum。
  • 然后启动 3 个 broker，注册到控制器。
• 优势：
  • 高可用：3 控制器 quorum 容忍 1 个故障；3 broker 支持分区复制。
  • 可扩展：易添加更多 broker，而不影响控制器。
  • KRaft 模式：取代 ZooKeeper，使用内置 Raft 协议，更轻量。
• 潜在问题与改进建议：
  • 无持久化：默认数据在容器重启丢失。建议添加 volumes。
  • 资源限制：未设置 CPU/Memory limits，高负载下可能 OOM。生产加 deploy.resources。
  • 日志：Kafka 日志输出到 stdout，可用 logging 配置。

启动集群

oot@ser745692301841:/dev_dir/kafka# ls
docker-compose.yml
root@ser745692301841:/dev_dir/kafka# docker-compose up -d
Creating controller-2 ... done
Creating controller-1 ... done
Creating controller-3 ... done
Creating broker-2     ... done
Creating broker-1     ... done
Creating broker-3     ... done
root@ser745692301841:/dev_dir/kafka# docker ps -a
CONTAINER ID   IMAGE                COMMAND                  CREATED         STATUS         PORTS                     NAMES
0956bdcbb17f   apache/kafka:4.1.0   "/__cacert_entrypoin…"   5 seconds ago   Up 4 seconds   0.0.0.0:49092->9092/tcp   broker-3
8e4c92bfc02f   apache/kafka:4.1.0   "/__cacert_entrypoin…"   5 seconds ago   Up 4 seconds   0.0.0.0:29092->9092/tcp   broker-1
4c4d32f110c9   apache/kafka:4.1.0   "/__cacert_entrypoin…"   5 seconds ago   Up 4 seconds   0.0.0.0:39092->9092/tcp   broker-2
ec7e79ab7a37   apache/kafka:4.1.0   "/__cacert_entrypoin…"   5 seconds ago   Up 5 seconds   9092/tcp                  controller-3
350c0ba616b7   apache/kafka:4.1.0   "/__cacert_entrypoin…"   5 seconds ago   Up 5 seconds   9092/tcp                  controller-1
8842b034516f   apache/kafka:4.1.0   "/__cacert_entrypoin…"   5 seconds ago   Up 5 seconds   9092/tcp                  controller-2

关闭集群

root@ser745692301841:/dev_dir/kafka# docker-compose down -v
Stopping broker-3     ... done
Stopping broker-1     ... done
Stopping broker-2     ... done
Stopping controller-3 ... done
Stopping controller-1 ... done
Stopping controller-2 ... done
Removing broker-3     ... done
Removing broker-1     ... done
Removing broker-2     ... done
Removing controller-3 ... done
Removing controller-1 ... done
Removing controller-2 ... done
Removing network kafka_default
root@ser745692301841:/dev_dir/kafka# docker ps -a
CONTAINER ID   IMAGE     COMMAND   CREATED   STATUS    PORTS     NAMES

进入一个broker

root@ser745692301841:/dev_dir/kafka# docker ps -a
CONTAINER ID   IMAGE                COMMAND                  CREATED          STATUS          PORTS                     NAMES
d08c34dd951f   apache/kafka:4.1.0   "/__cacert_entrypoin…"   18 seconds ago   Up 17 seconds   0.0.0.0:39092->9092/tcp   broker-2
66fd553b04b4   apache/kafka:4.1.0   "/__cacert_entrypoin…"   18 seconds ago   Up 17 seconds   0.0.0.0:29092->9092/tcp   broker-1
3e416ba03096   apache/kafka:4.1.0   "/__cacert_entrypoin…"   18 seconds ago   Up 17 seconds   0.0.0.0:49092->9092/tcp   broker-3
7edffc89670e   apache/kafka:4.1.0   "/__cacert_entrypoin…"   19 seconds ago   Up 18 seconds   9092/tcp                  controller-3
4aea596b68c7   apache/kafka:4.1.0   "/__cacert_entrypoin…"   19 seconds ago   Up 18 seconds   9092/tcp                  controller-2
d15a99927a5b   apache/kafka:4.1.0   "/__cacert_entrypoin…"   19 seconds ago   Up 18 seconds   9092/tcp                  controller-1
root@ser745692301841:/dev_dir/kafka# docker exec --workdir /opt/kafka/bin/ -it broker-1 sh
/opt/kafka/bin $ ls
connect-distributed.sh        kafka-client-metrics.sh          kafka-consumer-groups.sh     kafka-features.sh         kafka-metadata-quorum.sh       kafka-server-start.sh               kafka-streams-groups.sh             trogdor.sh
connect-mirror-maker.sh       kafka-cluster.sh                 kafka-consumer-perf-test.sh  kafka-get-offsets.sh      kafka-metadata-shell.sh        kafka-server-stop.sh                kafka-topics.sh                     windows
connect-plugin-path.sh        kafka-configs.sh                 kafka-delegation-tokens.sh   kafka-groups.sh           kafka-producer-perf-test.sh    kafka-share-consumer-perf-test.sh   kafka-transactions.sh
connect-standalone.sh         kafka-console-consumer.sh        kafka-delete-records.sh      kafka-jmx.sh              kafka-reassign-partitions.sh   kafka-share-groups.sh               kafka-verifiable-consumer.sh
kafka-acls.sh                 kafka-console-producer.sh        kafka-dump-log.sh            kafka-leader-election.sh  kafka-replica-verification.sh  kafka-storage.sh                    kafka-verifiable-producer.sh
kafka-broker-api-versions.sh  kafka-console-share-consumer.sh  kafka-e2e-latency.sh         kafka-log-dirs.sh         kafka-run-class.sh             kafka-streams-application-reset.sh  kafka-verifiable-share-consumer.sh

生产消费测试

创建topic

/opt/kafka/bin $ ./kafka-topics.sh --bootstrap-server broker-1:19092,broker-2:19092,broker-3:19092 --create --topic test-topic

Created topic test-topic.

开始消费

/opt/kafka/bin $ ./kafka-console-consumer.sh --bootstrap-server broker-1:19092,broker-2:19092,broker-3:19092 --topic test-topic --from-beginning

再打开一个终端

./kafka-console-producer.sh --bootstrap-server broker-1:19092,broker-2:19092,broker-3:19092 --topic test-topic

在producer终端敲的内容，会被consumer终端消费到。

# producer终端
root@ser745692301841:/dev_dir/kafka# docker exec --workdir /opt/kafka/bin/ -it broker-1 sh
/opt/kafka/bin $ ./kafka-console-producer.sh --bootstrap-server broker-1:19092,broker-2:19092,broker-3:1909
2 --topic test-topic
>
>nihao
>hello world
>^C/opt/kafka/bin $ 

# consumer终端
/opt/kafka/bin $ ./kafka-console-consumer.sh --bootstrap-server broker-1:19092,broker-2:19092,broker-3:19092 --topic test-topic --from-beginning


nihao
hello world

如果重新 执行 kafka-console-consumer.sh test-topic --from-beginning 将会看到 nihao、hello world 又被消费一遍，这是因为--from-beginning的原因，从头开始消费流内容。

主机硬件资源有限，我们可以用 一个 controller、一个broker

services:
  controller-1:
    image: apache/kafka:4.1.0
    container_name: controller-1
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: controller
      KAFKA_LISTENERS: CONTROLLER://:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:9093
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0

  broker-1:
    image: apache/kafka:4.1.0
    container_name: broker-1
    ports:
      - 29092:9092
    environment:
      KAFKA_NODE_ID: 4
      KAFKA_PROCESS_ROLES: broker
      KAFKA_LISTENERS: 'PLAINTEXT://:19092,PLAINTEXT_HOST://:9092'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://broker-1:19092,PLAINTEXT_HOST://localhost:29092'
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:9093
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
    depends_on:
      - controller-1

生态系统

Ecosystem

除了主发行版之外，还有大量工具可以与 Kafka 集成。生态系统页面 https://cwiki.apache.org/confluence/x/Ri3VAQ 列出了其中许多工具，包括流处理系统、Hadoop 集成、监控和部署工具。

升级

Upgrading

一些漏洞，以及版本之间的重大更新，与问题修复，详细内容去看官方文档。

KRaft与ZooKeeper

KRaft vs ZooKeeper

ZooKeeper 模式和 KRaft 模式之间存在许多差异。KRaft模式与 ZooKeeper 模式之间的差异页面列出了所有这些差异，包括配置、指标和行为变化。

https://kafka.apache.org/41/documentation/zk2kraft.html

不过入门学习，还没到不学这里不行，建议直接跳过，不要在此钻牛角尖。

兼容性

Compatibility

Kafka 4.0 的发布引入了重大变更，影响了各个组件之间的兼容性。为了帮助用户规划升级并确保无缝互操作性， 我们准备了 一个全面的 兼容性矩阵 https://kafka.apache.org/41/documentation/compatibility.html。

4.0 是一个重要里程碑，建议直接从 4.0 以上开始学习。

Docker

Docker

我们在 快速学习 Docker Compose搭建Kafka集群环境 部分已经有了些了解。

APIS

APIS

Kafka包含5个核心API：

1. Producer API 允许应用程序将数据流发送到Kafka 集群中的主题。
2. Consumer API 允许应用程序从 Kafka 集群中的主题读取数据流。
3. Streams API 允许将数据流从输入主题转换为输出主题。
4. Connect API 允许实现连接器，该连接器不断地从某些源系统或应用程序拉入 Kafka，或从 Kafka 推送到某些接收系统或应用程序。
5. Admin API 允许管理和检查主题、代理和其他Kafka对象。
6. Share Consumer API (Preview)

我喜欢node.js，所以用kafkajs来学习。

生产者API（Producer API）、消费者API（Consumer API）、分享消费者API Share Consumer API（Preview）、流API（Streams API）、连接API（Connect API）、管理API（Admin API）

配置

Configuration，在Kafka的官方文档中，整个文档界面内容的接近一半都是 “配置”部分的文档内容。这东西我们根本没法记住，只能多了解，在实践中了解些Kafka的一些特性 调优等等。

• 代理配置 Broker Configs
• 主题配置 Topic Configs
• 群组配置 Group Configs
• 生产者配置 Producer Configs
• 消费者配置 Consumer Configs
• Kafka Connect配置 Kafka Connect Configs
• 源连接器配置 Source Connector Configs
• 接收器连接器配置 Sink Connector Configs
• Kafka Streams配置 Kafka Stream Configs
• AdminClient配置 AdminClient Configs
• MirrorMaker配置 MirrorMaker Configs
• 系统属性 System Properties
• 分层存储配置 Tiered Storage Configs
• 配置提供程序 Configuration Providers
• 使用配置提供程序 Using Configuration Providers
• DirectoryConfigProvider DirectoryConfigProvider
• 环境变量配置提供程序 EnvVarConfigProvider
• 文件配置提供程序 FileConfigProvider
• Example:Referencing Files

Kafka支持提供了调整配置的方式给用户，但不一定需要用到大部分都是默认参数即可，只有我们需要定制化的时候再看这些配置内容也不迟。

设计

Design

动机

Motivation

我们设计 Kafka 的目的是让它成为一个统一的平台，处理大型公司可能拥有的 所有实时数据。为了实现这一点，我们必须考虑一系列相当广泛的用例。

它必须具有高吞吐量来支持大量事件流，例如实时日志聚合。

它需要妥善处理大量数据积压，以便能够支持来自离线系统的定期数据加载。

这也意味着系统必须处理低延迟传递才能处理更传统的消息传递用例。

我们希望支持对这些数据流进行分区、分布式、实时处理，从而创建新的派生数据流。这催生了我们的分区和消费者模型。

最后，在将流输入到其他数据系统进行服务的情况下，我们知道系统必须能够在机器故障时保证容错能力。

为了支持这些用途，我们设计了一个包含许多独特元素的系统，它更类似于数据库日志，而非传统的消息系统。

持久性

Persistence

Don’t fear the filesystem!, Kafka严重依赖文件系统来存储和缓存消息，人们普遍认为“磁盘很慢”，不要害怕，它可能比你想得块，没有你想得那么慢。

这段话的核心思想可以一句话概括为：

👉 别害怕用文件系统，顺序写 + OS 缓存比你想象的快。

简要总结如下：

1. 磁盘并不一定慢 —— 顺序读写速度可达数百 MB/s，比随机写快上千倍。
2. 现代操作系统很聪明 —— 会自动用所有空闲内存做磁盘缓存（page cache），预读、合并写入、提升性能。
3. 不用自己造缓存轮子 —— JVM 内存对象占用大且会触发 GC，直接依赖 OS 的缓存更高效。
4. Kafka 的设计理念 —— 把数据立即写入文件系统（即写入 OS page cache），而不是先放内存里再刷盘，这样既快又简单，还能利用系统级缓存持久化优势。

Constant Time Suffices.

Kafka 为什么不用复杂的数据结构（比如 BTree），而是选择简单的顺序写文件的设计。下面是通俗总结👇

1. 传统消息系统的做法

   • 一般会给每个消费者维护一个队列，用 BTree（或类似结构）来记录消息元数据。
   • BTree 的操作复杂度是 O(log N)，听起来不错，但对磁盘来说其实很慢。

2. 为什么 BTree 慢？

   • 每次访问树节点都可能导致一次 磁盘寻道（Disk seek）。
   • 一次磁盘寻道大概要 10 毫秒，而且一块盘一次只能做一个寻道。
   • 所以数据量一大、节点一多，性能就会急剧下降，比理论上的 log N 要慢得多。

3. Kafka 的思路：O(1) 就够了

   • Kafka 不用树，而是直接把消息顺序写入文件。
   • 每次写都是 append（追加）操作，读的时候也是顺序读。
   • 这种方式的操作是 O(1)（常数时间），不会因为数据量变大而变慢。
   • 读写互不影响，性能也更稳定。

4. 带来的好处

   • 可以用便宜的大容量机械硬盘（几TB那种），速度也不错。
   • 因为顺序读写不怕磁盘慢，只要带宽够就行。
   • 存储空间几乎无限，就可以长时间保留消息（比如保存一周）。
   • 消费者可以随时重读旧消息，而不需要立刻删除它们。

Kafka 的高性能来自于：

👉 “不玩花活”——不用复杂的 BTree、索引或事务系统，而是直接顺序写文件，让所有操作都接近常数时间，从而既快又省钱。

效率

Efficiency

这段主要讲 Kafka 为什么这么快，精简总结如下👇

1. 批量传输（Batching）

   • 把多条消息打包成一个“message set”，一起发送、写入、读取。
   • 这样减少网络往返次数和磁盘小 I/O 操作，写入变成顺序线性写，大幅提速。

2. 零拷贝（Zero-copy）

   • Kafka 的消息格式在生产者、Broker、消费者之间统一，数据可以直接复用。
   • 使用 Linux 的 sendfile 系统调用，让操作系统直接从文件缓存（page cache）发到网卡，不经过用户态。
   • 这样从“文件 → 内核 → 网络”只需要一次拷贝，极大减少 CPU 和内存开销。

3. 效果

   • 消费者能几乎以网络带宽上限速度读取消息。
   • 当消费者进度跟上时，Kafka 甚至不会真的访问磁盘（全在缓存里）。
   • 若启用 SSL，则不能用 sendfile，效率会稍低。

Kafka 靠 批量发送 + 零拷贝传输，把磁盘和网络效率压榨到极致，实现了超高吞吐量。

生产者

The Producer

Kafka Producer（生产者） 的工作方式

1. 负载均衡（Load Balancing）

• 生产者直接把消息发给对应分区（partition）的 leader broker，没有中间层。
• 所有 broker 都能提供元数据（哪台机器是哪个分区的 leader）。
• 分区策略：
  • 可随机分配（负载均衡）
  • 可按“键”（key）哈希分区，例如用用户 ID，让同一用户的数据都进同一个分区，方便消费者做本地化处理。

2. 异步发送与批量（Asynchronous Send + Batching）

• Producer 会先在内存里缓存多条消息，然后一次性批量发送。
• 可以设置最大消息数或最大等待时间（如 64KB 或 10ms）。
• 这样能大幅提升吞吐量，用少量延迟换取高效率。

Kafka Producer 通过 智能分区 + 异步批量发送 实现高效、可控的消息分发。

消费者

The Consumer

Kafka 消费者通过向其想要消费的分区所在的 Broker 发出“fetch”请求来工作。消费者在每次请求中指定其在日志中的偏移量，并从该位置开始接收一个日志块。因此，消费者对该位置拥有强大的控制权，并且可以根据需要回退该位置以重新消费数据。

Push vs. pull

为什么选择 消费者从Broker拉的方式，如果消费者消费速度较低，Broker一直推其实本质上成了一种拒绝服务攻击。拉取模式，Broker没数据消费者就进入了一种死循环，实际上就是忙着等待数据到达。为了避免这种情况，我们在拉取请求中设置了一些参数，允许消费者请求阻塞在“长轮询”中，直到数据到达。

Consumer Position

令人惊讶的是， 跟踪已消费的内容是消息传递系统的关键性能点之一。 大多数消息系统都会在 Broker 上保存已消费消息的元数据。也就是说，当消息被分发给消费者时，Broker 要么立即在本地记录该消息，要么等待消费者的确认。这是一个相当直观的选择，而且对于单机服务器来说，并不清楚这些状态还能保存在哪里。由于许多消息系统用于存储的数据结构扩展性较差，因此这也是一个务实的选择——由于 Broker 知道哪些消息已被消费，它可以立即删除，从而保持较小的数据大小。

或许不太明显的是，让 Broker 和 Consumer 就已消费的内容达成一致并非易事。如果 Broker 每次通过网络发送消息时都立即将其记录为 已消费，那么如果 Consumer 未能处理该消息（例如由于崩溃、请求超时或其他原因），该消息就会丢失。为了解决这个问题，许多消息系统添加了确认功能，这意味着消息在发送时仅被标记为 已发送，而不是 已消费；Broker 等待 Consumer 的特定确认，然后才将消息记录为 已消费。这种策略解决了消息丢失的问题，但也带来了新的问题。首先，如果 Consumer 处理了消息但在发送确认之前失败，那么该消息将被消费两次。第二个问题与性能有关，现在 Broker 必须为每条消息保留多个状态（首先锁定消息以防止再次发出，然后将其标记为永久消费以便将其删除）。必须处理一些棘手的问题，例如如何处理已发送但从未确认的消息。

Kafka 对此的处理方式有所不同。我们的主题被划分为一组完全有序的分区，每个分区在任何给定时间都只能被订阅消费者组中的一位消费者消费。这意味着消费者在每个分区中的位置只是一个整数，即下一条待消费消息的偏移量。这使得已消费消息的状态信息非常小，每个分区只有一个数字。此状态可以定期进行检查点。这使得消息确认的等效操作非常便宜。

这种做法还有一个附带好处。消费者可以故意回退到旧的偏移量并重新消费数据。这违反了队列的通用约定，但对许多消费者来说却是一项必不可少的功能。例如，如果消费者代码中存在 bug，并且在消费某些消息后才被发现，那么消费者可以在 bug 修复后重新消费这些消息。

Offline Data Load

可扩展的持久性允许消费者仅定期消费，例如批量数据加载，定期将数据批量加载到离线系统（如 Hadoop 或关系数据仓库）中。

Static Membership

这段话讲的是 Kafka 的「静态成员（Static Membership）」机制，下面是通俗易懂的总结👇

🧩 背景问题

在旧版本 Kafka 里，消费者组（Consumer Group）成员的身份是“临时的”：

• 每次消费者重启（比如升级、改配置、重启服务）后，它会重新加入组。
• 因为它的 ID 会变，Kafka 会以为是新成员加入 / 旧成员离开，于是触发 rebalance（重新分配分区）。
• 对大规模、带状态的应用来说，这会导致任务被重新分配、状态重建、性能下降，甚至暂时不可用。

💡 静态成员（Static Membership）的作用

为了解决上面的问题，Kafka 引入了 静态成员机制：

• 每个消费者实例都有一个固定 ID（group.instance.id）。
• 重启后 Kafka 还能认出它是同一个成员，不会触发 rebalance。
• 这样就能避免频繁的分区重新分配，提高可用性。

⚙️ 使用方法

1. Kafka 版本需 ≥ 2.3（Broker 和客户端都要）。

2. 给每个消费者实例配置一个唯一的：

   group.instance.id = your_unique_id

3. Kafka Streams 应用也是一样，每个实例设一个独立 ID。

4. 如果 ID 重复，Broker 会踢掉冲突的客户端并报错（FencedInstanceIdException）。

Kafka 静态成员机制让消费者“有固定身份”，重启不再引起 rebalance，应用更稳定、更快恢复。

消息传递语义

Message Delivery Semantics

Kafka的消息传递语义（Message Delivery Semantics），也就是Kafka发送、存储、消费消息时，能保证到什么程度不丢消息、不重复消息。

Kafka（或任何消息系统）都有三种语义

1. At most once（至多一次），消息可能丢，但绝不会重复（发了就算，有可能没收到）
2. At least once（至少一次），消息不会丢，但可能重复收到（可能重复发几次，为了安全）
3. Exactly once（恰好一次），消息既不会丢，也不会重复（最理想，但实现最复杂）

从生产者（Producer）角度看

• Kafka 写入机制：消息被写入“分区日志”，只有当所有同步副本（ISR）都写入成功，才算“提交（committed）”。 一旦提交，只要还有一台副本存活，就不会丢消息。

• 早期（0.11 之前）的问题： 如果生产者发消息时网络断了，它不知道消息到底写成功没，只能重发，这样会产生重复消息。所以只能保证 “至少一次”。

• Kafka 0.11+ 改进：幂等生产（Idempotent Producer） Broker 会给每个生产者分配唯一 ID，并检查消息的序列号。 如果重发的是旧的序号，就不会重复写入。 这样就能做到“不重复写”。

• 事务性发送（Transactional Producer） 允许一次事务性地写入多个分区或主题：要么全部成功，要么全部回滚。这是实现“exactly-once”的关键。

从消费者（Consumer）角度看

消费者需要记住自己“读到哪里了”（offset）。这决定了消息会不会重复处理或漏处理：

1. 读完 → 提交 offset → 再处理
   • 👉如果在提交后崩溃，处理没完成，但 offset 已推进 → 会漏消息
   • 🔹 对应 At most once（至多一次）
2. 读完 → 处理 → 再提交 offset
   • 👉 如果在提交前崩溃，消息会被重处理 → 会重复
   • 🔹 对应 At least once（至少一次）
3. 读 Kafka → 处理后写回 Kafka（比如 Kafka Streams）
   • Kafka 会把 “offset 提交” 和 “处理结果写入” 放进同一个事务
   • 要么都成功，要么都失败 ✅ 实现 Exactly once（恰好一次）

Exactly Once 的实现条件

Kafka 要做到真正的“恰好一次”，需要：

1. 生产者开启幂等和事务支持
2. 消费者使用 read_committed 模式（只读提交事务的数据）
3. 如果写入外部系统（比如 HDFS），则要能让 offset 和数据一并保存，否则只能做到“至少一次”。

Kafka 默认保证“至少一次”；通过“幂等 + 事务 + read_committed”，可以做到“恰好一次”。

使用事务

Using Transactions

背景：为什么需要事务？

在 Kafka 中，消息可能会因为：

• 网络波动；
• 消费者或生产者宕机；
• 分区重平衡（rebalance） 等原因出现 “消息重复” 或 “消息丢失”。

为了保证“消息只处理一次”，Kafka 从 0.11.0.0 版本开始引入了 事务（Transactions）机制。

事务能确保多个操作（发送消息、更新消费者偏移量）要么全部成功，要么全部失败。

Kafka 事务与传统数据库事务的区别

Kafka 的事务与数据库那种「BEGIN…COMMIT」事务有点不一样：

所以说：Kafka 的事务其实是由 生产者“帮消费者”原子地提交 offset 实现的。

实现 Exactly-Once 的三大关键要素

1. 消费者分区独占: 每个分区同一时刻只会由一个消费者处理（Kafka Consumer Group 机制自动保证）。

2. 生产者事务发送 生产者开启事务后，能保证：

   • 多条消息原子发送；
   • 同时提交 offset。

3. 一对一关系：一个消费者 ↔︎ 一个生产者 这样当 rebalance（再平衡）发生时，逻辑简单、可靠。 （复杂的多对多模式可以实现，但非常麻烦。）

消费者配置要求

想要实现 “exactly-once”，消费者要这样配置👇：

isolation.level=read_committed   # 只读取已提交的事务消息
enable.auto.commit=false         # 禁止自动提交 offset

如果你不设 read_committed，消费者会看到：

• 未完成的事务数据；
• 被中止（aborted）的事务消息。

这会导致数据重复或脏读。

生产者配置要求

生产者要指定一个 transactional.id：

transactional.id=my-tx-id

这会告诉 Kafka：

• “这个生产者要支持事务”；
• 如果应用重启，Kafka 会自动终止上次未完成的事务，避免重复。

生产者在发送时的伪代码：

producer.initTransactions();     // 初始化事务
producer.beginTransaction();     // 开始事务

// 1. 处理消息
// 2. 发送到目标 topic
// 3. 发送 offset 更新到 Kafka（表示处理完毕）

producer.commitTransaction();    // 全部成功，一起提交
// producer.abortTransaction();  // 有异常，一起撤销

异常与错误处理机制

Kafka 在新版本中对事务错误分类更细了，方便开发者处理：

事务中止后的恢复

如果事务中止（abort）：

• Kafka 会写入「事务标记记录（transaction marker）」说明事务被放弃；
• 消费者的 offset 需要手动回退；
• 这样可以重新处理被回滚的消息。

有两种恢复策略：

1. 简单粗暴法： 销毁并重建 Producer 和 Consumer；
2. 优雅法： 调用 seek() 回退到上次提交的 offset，重新消费。

Kafka 的事务机制可以实现：

“消费消息 → 处理 → 产出新消息 → 提交偏移量” 这一整套过程的 原子性（Atomicity）。

也就是说：

• 要么所有步骤都成功；
• 要么全部无效、不影响系统；
• 不会重复处理，也不会丢数据。

举个例子：事务性消息复制器

比如我们做一个“消息复制器”：

• 从 topic-A 消费；
• 处理；
• 再发送到 topic-B。

使用事务后，无论是网络闪断还是程序崩溃， Kafka 都能保证：👉 不会出现 “重复消息” 或 “漏消息” 的情况。

分享群组

Share Groups

Share Group是什么

以前Kafka里，消费者只有一种群体：Consumer Group（消费者群体）

而现在Kafka 4.1多了一种新模式：Share Group（共享组）

简单来说：Share Group = 可以让多个消费者同时消费同一个分区里的消息的一种新模式

这解决了老机制中一个限制：一个分区在同一时间只能被一个消费者消费。

与传统 Consumer Group的区别

为什么要有Share Group

传统 consumer group 的“分区独占”机制虽然能保证顺序和负载均衡，但在一些场景下就不太够用了，比如：

• 一个分区的流量太高，一个消费者吃不消；
• 希望同一个分区的消息能被多个消费者协同处理；
• 想要“任务队列”式的处理逻辑（每条消息独立 ack）。

💡 所以 Share Group = Kafka 版的任务队列（Task Queue）模型 有点像 RabbitMQ 的「Work Queue」或 SQS 的「Visibility Timeout」。

Share Group 是怎么工作的？

可以这样想象：

• 多个消费者一起盯着同一个分区里的消息，
• 谁先“抢到锁（lock）”，谁就能消费这条消息。

核心机制：消息锁（Record Lock）

每当消费者从 Kafka 拉取消息时：

1. 消息会被“锁定”（acquired）给某个消费者；
2. 锁有时效（默认 30 秒）；
3. 在锁的有效期内，这条消息不会被其他消费者看到；
4. 锁到期自动释放，别的消费者就能接手。

🕒 锁时长可调：

share.record.lock.duration.ms=30000   # 默认 30 秒

消费者对消息可以做四种操作：

Kafka 还提供了一些限制和保护机制

Kafka Broker 会限制：

• 每个分区最多同时被多少条消息“锁住”，防止内存或处理过载。

配置项：

group.share.partition.max.record.locks

这个机制让系统能自动恢复，即使有消费者宕机、延迟，也不会卡死。

多个 Share Group 之间的关系

如果同一个 topic 被多个 share group 订阅：

• 每个 share group 都会独立消费整条 topic；
• 它们互不影响，就像多个独立的任务队列。

举个例子：

假设你有一个 topic 叫 image-jobs，每条消息是一张图片要处理。

• 旧模式下：

  • 分区数 4；
  • 只有 4 个消费者能同时工作；
  • 其中一个消费者崩溃，该分区的任务都停滞。

• 新模式（Share Group）：

  • 你可以有 20 个消费者；
  • 所有消费者都从分区里抢任务；
  • 每条任务加锁 30 秒；
  • 没处理完会自动重试；
  • 就像一个「分布式线程池」。

复制

Replication

这部分（Replication）讲的是 Kafka 的数据复制机制，也就是 Kafka 如何在多台服务器之间保证数据不丢、不停机。我来帮你用最通俗的方式总结👇

为什么要复制（Replication）

Kafka 每个 topic 都会被分成多个 分区（partition）。 每个分区的数据会在多台机器上复制（replicate）几份，比如复制因子为 3，代表有：

• 1 个主副本（leader）
• 2 个备份副本（followers）

👉 当 leader 挂了，Kafka 会自动切换到其中一个 follower，保证数据不丢、服务不中断。

复制机制是怎么工作的

• Leader 负责写入：所有消息都先写到 leader。
• Follower 负责同步：每个 follower 都从 leader 拉取数据，同步到自己的日志中。
• Kafka 会维护一个集合叫 ISR（In-Sync Replicas），表示“已经跟上 leader 的副本们”。
  • 只要还剩一个副本在 ISR 里，Kafka 就能保证数据不丢。

Kafka 怎么判断一个副本“还活着”

1. 定期心跳（heartbeat） —— 如果 follower 一段时间没汇报，controller 就认为它挂了。
2. 不能落后太多 —— 如果同步太慢（超出 replica.lag.time.max.ms），也会被踢出 ISR。

被踢出 ISR 的副本不能被选为 leader。

Kafka 的复制模型 vs 其他系统

Kafka 没采用像 Raft 那种“多数投票（majority quorum）”机制。 它用了更高效的 ISR 模型：

• 优点：只要 ISR 里的副本都同步成功，就认为“写入成功”，性能更高、磁盘占用更少。
• 缺点：可用性和一致性之间要取舍。

👉 简单理解：

Raft 类系统靠“投票”，Kafka 靠“跟得上的那群副本（ISR）”。

如果所有副本都挂了怎么办？

Kafka 有两种策略（通过 unclean.leader.election.enable 控制）：

Producer 写入时的可靠性选项

生产者可以设置 acks 参数控制写入确认级别：

另外可以通过：

min.insync.replicas = 2

要求至少有 2 个副本同步成功才能确认，进一步提高可靠性。

Leader 选举机制

• Kafka 集群里有一个特殊节点叫 Controller，负责管理哪些 broker 存活、谁当 leader。

• 当一个 broker 挂掉，controller 会：

  • 选出新的 leader（从 ISR 内部）；
  • 批量通知集群更新元数据；
  • 如果 controller 自己挂了，会重新选一个。

这样即使几百个分区都要选新 leader，Kafka 也能在几秒内恢复。

Kafka 的可靠性保证

✅ 不会丢已提交（committed）的消息，前提是：

• 至少有 1 个 ISR 副本存活。
• Producer 设置了 acks=all。

❌ 不能保证强一致性（比如同时断网、ISR 全挂的情况会暂停写入）。

Kafka 的复制机制靠：

• Leader + Follower + ISR 集合 + Controller 管理
• 实现了：
  • ✅ 高可靠（不丢数据）
  • ✅ 高可用（节点故障自动恢复）
  • ✅ 高性能（避免频繁投票的延迟）

日志压缩

Log Compaction

核心概念：什么是 Log Compaction？

Kafka 的 日志压缩（Log Compaction） 是一种“更聪明的日志保留机制” 👉 它确保 每个 key 的最后一条最新消息 永远不会被删除。

对比普通的日志清理方式：

• 普通保留（Retention）：按时间或大小清理旧消息。🧹 比如保存最近 7 天的消息。
• 日志压缩（Compaction）：按 key 保留每个 key 的最后一次更新。📦 比如用户 ID = 123 的最新邮箱记录。

举个例子

假设 topic 中存储用户邮箱变更：

123 => bill@microsoft.com
123 => bill@gatesfoundation.org
123 => bill@gmail.com

Kafka 会在压缩后只保留：

123 => bill@gmail.com

这意味着：

• 即使日志被清理，仍能恢复每个用户的“当前状态”。
• 不需要保存所有历史变化。

应用场景

1. 数据库变更同步（Change Data Capture）

   • 把数据库变化同步到缓存、搜索集群等系统。
   • 如果缓存丢了，可以通过压缩后的日志恢复完整数据。

2. 事件溯源（Event Sourcing）

   • 系统状态由一连串事件组成，压缩保证能重建最终状态。

3. 高可用日志（Journaling）

   • 服务崩溃后，可以重放压缩日志恢复到最新状态。

工作机制（简化版）

Kafka 在后台有一组 log cleaner 线程，它们做的事是：

1. 扫描日志文件；
2. 记录每个 key 的最新 offset；
3. 把旧的相同 key 的消息删除；
4. 生成一个“干净的”新日志段。

📌 过程是异步进行的，不会影响正常的读写。

🧱 保证与特点

🔧 如何启用

在 topic 配置中设置：

log.cleanup.policy=compact

还可以调节压缩的行为：

log.cleaner.min.compaction.lag.ms=60000  # 最少保留多久后才能被压缩
log.cleaner.max.compaction.lag.ms=3600000 # 最长多长时间后必须压缩

🧩 一句话总结

Kafka 的 Log Compaction 就像是一个“自动同步的数据库快照系统”，它让你能同时：

• 实时消费最新变更；
• 在任何时候恢复完整状态。

配额

Quotas

🎯 作用

Kafka 的配额系统是用来限制客户端使用 Broker 资源的速度，防止单个生产者或消费者过度占用资源，导致：

• 网络被占满；
• Broker CPU 忙不过来；
• 影响其他正常客户端（类似“限流”机制）。

⚙️ 两种主要配额类型

1. 网络带宽配额（Network Bandwidth Quotas）

   • 限制客户端每秒最多能发送/接收多少字节（bytes/sec）。
   • 超过后会被限速（throttled），Broker 会延迟响应。

2. 请求速率配额（Request Rate Quotas）

   • 限制客户端使用 Broker CPU/线程资源的比例。
   • 单位是 Broker I/O 线程和网络线程总资源的百分比（例如 5%）。

👥 配额作用对象（Client Groups）

Kafka 根据以下信息区分客户端：

• user（用户）
• client-id（客户端ID）

可以针对以下对象配置配额：

• (user, client-id) 组合（最精确）
• 仅 user
• 仅 client-id

Kafka 会优先使用最精确匹配的配额规则。

🏗️ 配置方式与优先级

Kafka 支持动态修改，不需要重启集群。

优先级顺序（从高到低）：

1. 匹配的 (user, client-id)
2. 匹配 user + 默认 client-id
3. 匹配 user
4. 默认 user + 匹配 client-id
5. 默认 user + 默认 client-id
6. 默认 user
7. 匹配 client-id
8. 默认 client-id

⚡ 触发机制（Enforcement）

当 Broker 检测到客户端超出限额时：

1. 计算需要延迟多久；
2. 返回一个带延迟时间的响应；
3. 暂时关闭与客户端的通道；
4. 等延迟结束后再恢复通信。

即使客户端版本旧、不理解“延迟”响应，Broker 也会通过“静音通道（muted channel）”强制限速。

实现细节

• 每个配额是按 Broker 计算的，而不是整个集群。
• 这样每台 Broker 自己控制限速，更易实现、开销小。
• 统计是基于多窗口滑动平均（如 30 个 1 秒窗口），以更平滑地检测异常。

💡 简单比喻

Kafka Quota 就像网吧管理员：

• 每个用户（或一组用户）都有带宽上限；
• 如果你占太多网速，系统会自动限速一会儿；
• 既防止“熊孩子”把网络塞爆，又能保障大家流畅使用。

实现

Implementation

网络层

Network Layer

网络层是一个相对简单的 NIO（非阻塞 I/O）服务器，因此这里不会进行太详细的描述。

• sendfile 优化： 通过为 TransferableRecords 接口增加一个 writeTo() 方法，使得基于文件的消息集（file-backed message set）可以直接使用操作系统提供的 transferTo() 调用。 👉 这样能跳过用户态缓冲区，直接从文件传输到网络，提高效率。

• 线程模型：

  • 一个 acceptor 线程：负责监听和接受新的连接。
  • 多个 processor 线程（N 个）：每个处理固定数量的连接。 这种设计在其他系统中已经被广泛验证，实现简单、性能优秀。

• 协议设计： Kafka 的网络通信协议被保持得非常简单， 这样做是为了 便于将来用其他语言实现客户端（比如 Java 以外的语言）。

📘 简单理解：

Kafka 的网络层类似于：

• 一个主线程负责「接客」；
• 一群工作线程负责「服务」；
• 文件内容发送时不走用户态缓存（直接走操作系统通道）；
• 协议尽量简单，方便跨语言实现。

消息

Messages

消息由三个部分组成：

• 可变长度的头部（header），
• 可变长度的不透明键字节数组（opaque key byte array），
• 以及可变长度的不透明值字节数组（opaque value byte array）。

将键（key）和值（value）保持为“不透明”的设计是正确的决策——当前各种序列化库的发展非常迅速，而任何特定的序列化方式都不太可能适用于所有场景。显然，具体使用 Kafka 的某个应用通常会在自身规范中指定一种固定的序列化类型。

RecordBatch 接口只是一个用于遍历消息的迭代器，同时还提供了一些专门用于在 NIO 通道中批量读取和写入消息的方法。

消息格式

Message format

Kafka 中的消息（Message，也称 Record）并不是单独写入磁盘的，而是成批（batch）写入的。一个批次称为 RecordBatch，它包含一个或多个 Record（记录）。

整体结构关系

RecordBatch
 ├── Header（批次头）
 ├── RecordsCount（记录数）
 └── Records[]（每条消息）
      ├── Record Header
      ├── Key / Value
      └── Headers（消息头）

Record Batch（记录批）

Kafka 的核心写入单元。包含多条消息及其元信息。

🧾 结构字段

⚙️ Attributes（属性位解析）

🧮 CRC 校验说明

• 覆盖范围： 从 attributes 到批尾所有字节。
• 不包含： partitionLeaderEpoch 字段（避免 Broker 修改此字段时要重算 CRC）。
• 算法： CRC-32C (Castagnoli)。

♻️ Log Compaction（日志压缩影响）

压缩时：

• 保留 首尾 offset/sequence，以便 Producer 恢复状态。
• 即使所有消息被清理掉，批次也可能被保留（用于保持序列号连续）。
• baseTimestamp 可能会改变（若首条被清理或为删除标记记录）。
• 当包含“删除标记”或“事务中止标记”时，delete horizon 会更新。

⚡Control Batch（控制批）

控制批不是普通消息，用于管理事务提交/中止。

• 控制批中只含 一条记录（control record）。
• 应用程序消费者不应看到这些消息。
• 用途：让消费者过滤掉已中止的事务消息。

🧭 控制记录 Key 结构

Value 内容对客户端是不透明的（Kafka 内部使用）。

🧱Record（单条消息）

每条消息在批次中都有相对偏移、时间戳、键值等。

🧾 Record结构字段

🏷️Record Header（消息头）

每条记录可包含多个 header（键值对），便于携带附加元信息。

👉 使用 Protobuf 的 varint 编码格式（节省空间）。

🕰️旧版格式（Pre-0.11）

在 Kafka 0.11 之前，消息采用 Message Set 格式：

• 没有 RecordBatch 概念；
• 不支持事务与批压缩；
• 现已被 RecordBatch 替代。

🧠总结思维导图式归纳

Message (Record)
 └── RecordBatch
      ├── Header(批次元信息)
      │    ├── baseOffset / batchLength / crc ...
      │    └── attributes(压缩/事务/控制批等)
      ├── RecordsCount
      └── Records[]
           ├── timestampDelta / offsetDelta
           ├── key / value
           └── Headers[]
                ├── headerKey
                └── headerValue

💡核心要点回顾

• ✅ Kafka 以 RecordBatch 为最小写入单元；
• ✅ 支持多种压缩算法，节省磁盘与网络带宽；
• ✅ 事务消息通过 Control Batch 管理；
• ✅ CRC-32C 保证批数据完整性；
• ✅ Log Compaction 会保留首尾 offset/sequence；
• ✅ 每条消息的 key/value 都可能为空；
• ✅ header 使用 Protobuf varint 编码，灵活高效。

日志

Log

🧱主题与分区的日志目录结构

• 主题 my-topic 有 两个分区 ⇒ 生成两个目录：

my-topic-0/
my-topic-1/

• 每个目录中存放该分区的日志文件（log files），文件中记录该分区的所有消息。

📂日志文件格式

每个日志文件由连续的日志条目（log entries）组成：

[4字节: N] [N字节: message]

• N：当前消息的字节长度（int32）
• message：消息本体（N 个字节）

每条消息在文件中的位置由一个唯一的 offset（偏移量） 标识。

🔢消息偏移（Offset）

• 每条消息都有一个 64 位整数 offset。
• offset 表示该消息在分区数据流中起始的 字节位置。
• offset 单调递增、仅在分区内唯一。
• 消费者通过 offset 来标识和读取消息位置。

🗂️日志文件命名规则

• 每个日志文件的文件名 = 该文件中第一条消息的 offset。

00000000000000000000.log     # 第一批
00000000000000400000.log     # 下一批（假设 max log size 为 4MB）

• 当日志文件达到配置的最大大小 S（max log file size）后，会创建下一个文件。

⚙️记录格式版本化（Record Format Versioning）

• Kafka 的消息格式是 版本化（versioned） 的。
• 这样可以确保：
  • Producer、Broker、Consumer 之间传输时无需重新编码；
  • 不同组件可直接共享批数据（RecordBatch）；
  • 便于向后兼容和协议升级。

💡为何用 offset 而非 GUID 作为消息ID

Kafka 最初考虑过使用 GUID（全局唯一 ID），但最终选择了 offset：

Offset 本质上是“分区内的消息计数器”。因为 offset 在分区中唯一且递增，所以查找更简单、性能更高。

🚀设计取舍

• Offset 是 实现细节，对用户 API 是透明的；
• 但对 Kafka 内部（日志、存储、索引）来说极为关键；
• 用 offset 替代 GUID 让系统更轻量高效，即：
  • 简化了查找结构；
  • 减少磁盘索引同步开销；
  • 保持了消息唯一性。

🧭总结一句话

• Kafka 每个分区的日志文件就是一串有序的消息序列，
• 每条消息通过顺序递增的 offset 唯一标识，
• offset 既是消息位置，也是 Kafka 高效存储和消费的核心。

📘 Kafka 日志系统机制总结（Writes / Reads / Deletes / Guarantees）

Kafka 的日志（log）是 顺序写入、可顺序读取、可分段删除 的磁盘结构。它同时在性能与持久性之间做了平衡设计。

✍️Writes（写入机制）

🧱 写入规则

• 所有消息 顺序写入（append-only），始终写入最后一个日志文件；
• 当文件达到配置上限（如 1GB）时，滚动（roll） 创建新文件。

⚙️ 两个关键参数

📜 持久化保证：系统崩溃时，最多丢失 M 条消息 或 S 秒数据。

📖Reads（读取机制）

🧩 读取方式

• 客户端根据 64 位 offset（消息逻辑偏移）读取；
• 同时指定一个 S 字节的最大读取块大小（max chunk size）；
• 返回结果是一个消息迭代器（iterator），包含该块内所有消息。

⚠️ 异常情况

• 若单条消息超过缓冲区大小：
  • 客户端可重试，多次加倍 buffer 大小；
  • 或者服务端直接拒绝超出最大消息大小的请求。

🧮 读取步骤

1. 根据 offset 找到对应的 日志段文件（log segment）；
2. 计算该文件内的 相对偏移（file-specific offset）；
3. 从文件中读取该偏移开始的消息。

查找采用一种简化的 二分查找（binary search variation），基于每个段的内存索引范围来定位。

特殊读取场景

• 可以直接获取“最新消息”，即从“此刻”开始消费；
• 若消费者请求的 offset 不存在（过期或被删除）：
  • Broker 返回 OutOfRangeException；
  • 消费者可选择“重置 offset”或“停止消费”。

📦 读取结果格式

单个分区（MessageSetSend）

total length : 4 bytes
error code   : 2 bytes
message 1    : x bytes
...
message n    : x bytes

多分区（MultiMessageSetSend）

total length : 4 bytes
error code   : 2 bytes
messageSetSend 1
...
messageSetSend n

🗑️Deletes（删除机制）

Kafka 删除数据以“日志段（log segment）为单位”。

🕒 删除策略

⚙️ 并发删除机制

为避免锁住读取操作：

• Kafka 使用 Copy-on-Write（写时复制） 的段列表结构；
• 删除时创建新段列表快照；
• 读操作在静态快照上继续进行（二分查找依然安全）。

🛡️Guarantees（数据一致性与恢复机制）

💾 Flush 保证

• 参数 M 控制多少消息后强制刷盘；
• 崩溃时只会丢失 尚未刷盘的消息。

🔄启动恢复（Recovery）

重启时会执行 日志恢复过程：

1. 遍历最新的日志段；
2. 检查每条消息是否有效：
   • ✅ 条件1：offset + size < 文件长度
   • ✅ 条件2：消息体的 CRC32 校验通过
3. 若发现损坏（corruption）：
   • 截断（truncate）到最后一条合法消息。

⚠️需要防御的两类损坏

CRC 检查确保日志不会被伪数据污染，即使系统崩溃时元数据与数据写入顺序不一致。

🧠整体机制总结

💬一句话总结

• Kafka 的日志是一个可顺序写入、可顺序读取、分段存储的持久化结构，
• 通过 flush 控制、CRC 校验与 copy-on-write 删除机制，
• 在高性能与数据安全之间取得了理想的平衡。⚡

分布

Distribution

🧩分布式机制：消费者偏移量追踪（Consumer Offset Tracking）

Kafka 的消费者会追踪自己在每个分区中消费到的最大偏移量（offset），并且可以“提交”（commit）这些偏移量，以便在消费者重启后能从上次的位置继续消费。

Kafka 提供了一种机制，把某个消费者组的所有偏移量存储在一个特定的 broker（称为组协调者 group coordinator）上。 也就是说，一个消费者组中的所有消费者实例，都必须把它们的 offset 提交（commit）和读取（fetch）请求发送给同一个组协调者。

Kafka 根据消费者组的名称来决定哪个 broker 充当这个组的协调者。 消费者可以向任意一个 broker 发送 FindCoordinatorRequest 请求来查找自己的协调者，然后从 FindCoordinatorResponse 里获取协调者的具体信息。 之后，消费者就可以直接向这个协调者 broker 提交或读取偏移量。 如果协调者发生变化（比如挂掉或重新选举），消费者需要重新查找新的协调者。

偏移量提交可以由消费者自动提交（auto commit）或手动提交（manual commit）。

当组协调者收到一个 OffsetCommitRequest 时，它会把请求写入一个特殊的 Kafka 内部主题 —— __consumer_offsets。 只有当该偏移量消息被该主题的所有副本都成功接收后，broker 才会向消费者返回提交成功的响应。

如果在可配置的超时时间内，这些副本没有同步成功，那么提交就会失败，消费者可能会在等待一段时间后重试。

Kafka 会定期对 __consumer_offsets 主题进行压缩（compaction），因为它只需要保留每个分区最新的偏移量记录。 同时，协调者会把这些偏移量缓存在内存中，以便快速响应消费者的读取请求。

当协调者收到一个 OffsetFetchRequest 请求时，它会直接从内存缓存中返回最近提交的偏移量。 但如果协调者刚刚启动，或者刚成为某个消费者组的新协调者（例如它现在负责的 offsets 主题分区刚切换到它），此时缓存还没加载完成。 在这种情况下，协调者会返回一个 CoordinatorLoadInProgressException 异常，表示“正在加载”，消费者需要稍等一会再重试。

💡通俗解释

你可以把 Kafka 的偏移量提交机制理解为一种“断点续播记录”：

• 消费者每读一条消息，就会记下“我读到第几条了”。
• 这个“进度记录”会提交给一个专门的“记录员”（组协调者）。
• 所有同组的消费者都找同一个“记录员”汇报。
• 记录员把大家的进度写到一个隐藏的系统账本 __consumer_offsets。
• Kafka 会定期整理账本，只保留最新的那一条“进度记录”。
• 当消费者重启时，它就能问协调者：“我上次读到哪了？”然后从那里继续消费。

⚙️关键点总结

KafkaJS For Node.js

KafkaJS 官方自己的介绍是，A modern Apache Kafka client for Node.js。一个现代的Kafka Node.js客户端。

KafkaJS-Intro to Kafka

Kafka是一个消息系统，可以在系统之间安全地移动数据。根据每个组件的配置方式，它可以充当实时事件跟踪的传输器，也可以充当复制的分布式数据库。虽然它通常被称为队列，但更准确地说，它介于队列和数据库之间，兼具两种系统的特性和优势。

KafkaJS-Glossary

KafkaJS-Message Formats

虽然我们通常将主题中的数据称为“消息”，但消息并没有统一的格式。从 Kafka 的角度来看，消息只是一个键值对，其中键和值都是字节序列。数据生产者和消费者需要就其格式达成一致。通常，您会找到纯文本的无模式消息，例如 JSON，或者强制模式的二进制格式，例如 AVRO。

纯文本JSON，JSON 无需介绍。它简单易用。我们唯一需要做的就是将消息转换Buffer为字符串并进行解析，使用 JSON 的缺点是它不强制任何类型的模式，因此在解析消息后，您无法知道哪些字段可用以及它们的类型。数据生产者无法保证字段一定会存在或其类型不会改变，这使得处理起来非常困难且容易出错。

AVRO、Protobuf 之类的数据化序列工具也可以。

KafkaJS-Getting Started

docker-compose.yml

services:
  controller-1:
    image: apache/kafka:4.1.0
    container_name: controller-1
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: controller
      KAFKA_LISTENERS: CONTROLLER://:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:9093
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs  # 可选：指定日志目录
      TZ: UTC

  broker-1:
    image: apache/kafka:4.1.0
    container_name: broker-1
    ports:
      - 29092:29092
    environment:
      KAFKA_NODE_ID: 2
      KAFKA_PROCESS_ROLES: broker
      KAFKA_LISTENERS: 'PLAINTEXT://:19092,PLAINTEXT_HOST://:29092'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://broker-1:19092,PLAINTEXT_HOST://127.0.0.1:29092'
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:9093
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      # 新增：修复单节点内部主题副本问题
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs  # 可选：指定日志目录
      TZ: UTC
    depends_on:
      - controller-1

一个简单的demo

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
    clientId: 'my-app',// 客户端ID
    // Kafka集群地址,把Broker的地址写到数组内，每个item为一个Broker
    brokers: ['127.0.0.1:29092']
});

const create_topic = async () => {
    const admin = kafka.admin();
    await admin.connect();

    const topics = await admin.listTopics();

    console.log("已有的Topic: ", topics);

    if (!topics.includes('test-topic')) {
        console.log("创建Topic: test-topic");
        const topicConfig = {
            topics: [
                {
                    topic: 'test-topic',//主题名
                    numPartitions: 1,//分区数
                    replicationFactor: 1//副本数
                }],
            waitForLeaders: true//等待leader分配完成
        };

        const result = await admin.createTopics(topicConfig);
        console.log('topic created:', result);
    }

    console.log("已有的Topic: ", await admin.listTopics());

    await admin.disconnect();
};

// 创建一个生产者向主题生成消息
const producer_run = async () => {
    const producer = kafka.producer();
    await producer.connect();
    const send_result = await producer.send({
        topic: 'test-topic',
        messages: [
            { value: `Hello KafkaJS user! ${Date.now()}` },
        ],
    });

    console.log('send result:', send_result);
    if (send_result && send_result.length > 0) {
        console.log(`消息发送成功，主题: ${send_result[0].topicName}, 分区: ${send_result[0].partition}, 偏移量: ${send_result[0].baseOffset}`);
    } else {
        console.log('消息发送失败');
    }

    await producer.disconnect();
};

// 创建一个消费者从主题消费消息
const consumer_run = async () => {

    let consumeCounter = 0;

    const consumer = kafka.consumer({ groupId: 'test-group' });
    await consumer.connect();
    await consumer.subscribe({ topic: 'test-topic', fromBeginning: true });

    await consumer.run({
        eachMessage: async ({ topic, partition, message }) => {
            consumeCounter++;
            console.log(`消费消息: consumeCounter[${consumeCounter}]`, {
                value: message.value.toString(),
                topic,
                partition
            })
        },
    });

    console.log("消费者已启动，等待消息...");
};

(async () => {
    await create_topic();

    await consumer_run();

    setInterval(() => {
        producer_run();
    }, 1000);

})();

可以观察到每秒都会生产一条消息到 Topic test-group 中，消费者接收到消息。

root@ser745692301841:/dev_dir/kafka-demo/kafka/kafkajs# node index.js
(node:173874) TimeoutNegativeWarning: -1761297030774 is a negative number.
Timeout duration was set to 1.
(Use `node --trace-warnings ...` to show where the warning was created)
已有的Topic:  [ 'test-topic', '__consumer_offsets' ]
已有的Topic:  [ 'test-topic', '__consumer_offsets' ]
{"level":"INFO","timestamp":"2025-10-24T09:10:30.820Z","logger":"kafkajs","message":"[Consumer] Starting","groupId":"test-group"}
{"level":"INFO","timestamp":"2025-10-24T09:10:30.865Z","logger":"kafkajs","message":"[ConsumerGroup] Consumer has joined the group","groupId":"test-group","memberId":"my-app-b32f3f97-93b4-4bcb-8292-e3a959f26640","leaderId":"my-app-b32f3f97-93b4-4bcb-8292-e3a959f26640","isLeader":true,"memberAssignment":{"test-topic":[0]},"groupProtocol":"RoundRobinAssigner","duration":42}
消费者已启动，等待消息...
{"level":"WARN","timestamp":"2025-10-24T09:10:31.868Z","logger":"kafkajs","message":"KafkaJS v2.0.0 switched default partitioner. To retain the same partitioning behavior as in previous versions, create the producer with the option \"createPartitioner: Partitioners.LegacyPartitioner\". See the migration guide at https://kafka.js.org/docs/migration-guide-v2.0.0#producer-new-default-partitioner for details. Silence this warning by setting the environment variable \"KAFKAJS_NO_PARTITIONER_WARNING=1\""}
send result: [
  {
    topicName: 'test-topic',
    partition: 0,
    errorCode: 0,
    baseOffset: '103',
    logAppendTime: '-1',
    logStartOffset: '0'
  }
]
消息发送成功，主题: test-topic, 分区: 0, 偏移量: 103
消费消息: consumeCounter[1] {
  value: 'Hello KafkaJS user! 1761297031874',
  topic: 'test-topic',
  partition: 0
}
send result: [
  {
    topicName: 'test-topic',
    partition: 0,
    errorCode: 0,
    baseOffset: '104',
    logAppendTime: '-1',
    logStartOffset: '0'
  }
]
消息发送成功，主题: test-topic, 分区: 0, 偏移量: 104
消费消息: consumeCounter[2] {
  value: 'Hello KafkaJS user! 1761297032873',
  topic: 'test-topic',
  partition: 0
}
^C

KafkaJS-Client Configuration

客户端必须至少配置一个broker,在列表中的broker被作为种子broker用于引导客户端和加载初始元数据。

const { Kafka } = require('kafkajs')

// Create the client with the broker list
const kafka = new Kafka({
    clientId: 'my-app',
    brokers: ['broker1:9092', 'broker2:9092']
});

KafkaJS-Client Id

应用程序的逻辑标识符，Broker可以使用它来对特定应用程序应用配额或跟踪请求，如 booking-events-processor。

kafka文档将clientId描述为，clientId是客户端的逻辑分组，其名称由客户端应用程序选择，且具有意义的名称。元组(user,clientId)定义了一个安全的客户端逻辑组，这些客户端共享用户主题和客户端ID。 配额可以应用于(user,client-id)、用户组或客户端ID组。

官方文档还说，clientId 发出请求时传递给服务器的ID字符串，通过允许在服务器端请求日志中包含逻辑应用程序名称，从而能追踪请求来源（而不仅仅是IP和端口）

因此，clientId应该在集群或水平扩展的应用程序中跨多个实例共享，但每个应用程序都不同。但是每个客户端进程尽可能都用独立的ClientID 应用+实例ID组成字符串。有些情况能共享有些不能尽量一刀切。

KafkaJS-Broker discovery

通常情况下，KafkaJS会自动感知到Broker集群拓扑结构的变化并做出相应，但在某些情况下，你可能希望能够动态获取种子 Broker，而不是使用静态配置的列表，这种情况下，brokers可以被设置为解析为Broker数组的异步函数。

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: async () => {
    // Example getting brokers from Confluent REST Proxy
    const clusterResponse = await fetch('https://kafka-rest:8082/v3/clusters', {
      headers: 'application/vnd.api+json',
    }).then(response => response.json())
    const clusterUrl = clusterResponse.data[0].links.self

    const brokersResponse = await fetch(`${clusterUrl}/brokers`, {
      headers: 'application/vnd.api+json',
    }).then(response => response.json())

    const brokers = brokersResponse.data.map(broker => {
      const { host, port } = broker.attributes
      return `${host}:${port}`
    })

    return brokers
  }
});

此发现机制仅用于获取初始 Broker 集合（即种子Broker）。成功连接到此列表中的Broker后，Kafka会通过其自身的机制来发现集群中的其余Broker。

KafkaJS-SSL

该ssl选项可用于配置 TLS 套接字。这些选项将直接传递给 TLS 安全上下文tls.connect并用于创建 TLS 安全上下文，所有选项均可接受。

const fs = require('fs')

new Kafka({
  clientId: 'my-app',
  brokers: ['kafka1:9092', 'kafka2:9092'],
  ssl: {
    rejectUnauthorized: false,
    ca: [fs.readFileSync('/my/custom/ca.crt', 'utf-8')],
    key: fs.readFileSync('/my/custom/client-key.pem', 'utf-8'),
    cert: fs.readFileSync('/my/custom/client-cert.pem', 'utf-8')
  },
})

NODE_EXTRA_CA_CERTS可用于添加自定义 CA。ssl: true如果您没有任何额外配置并希望启用 SSL，请使用此选项。

如果Kafka broker只启用了SSL（不要求客户端证书），可以这样写

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['broker-ssl.mydomain.com:9093'],
  ssl: true  // 简写形式，默认使用 Node.js 系统信任的 CA
});

Kafka中SSL常见场景

KafkaJS-SASL

在Kafka中，SASL（Simple Authentication and Security Layer）是一种 身份验证机制框架，用于在客户端（如生产者、消费者、管理工具等）与Kafka Broker之间进行安全认证。

简单来说：SASL是“你是谁”的验证机制，它让Kafka在接受请求之前，确认客户端的身份是否合法。

KafkaJS支持 PLAIN、SCRAM-SHA-256、SCRAM-SHA-512和AWS机制。

请注意，如果您未使用 TLS，即使凭证本身有效，代理也可能配置为拒绝您的身份验证尝试。特别提醒，在使用 TLSPLAIN作为身份验证机制时，切勿在不使用 TLS 的情况下进行身份验证，因为这会以未加密的纯文本形式传输您的凭证。

Options

PLAIN与SCRAM示例

new Kafka({
  clientId: 'my-app',
  brokers: ['kafka1:9092', 'kafka2:9092'],
  // authenticationTimeout: 10000,
  // reauthenticationThreshold: 10000,
  ssl: true,
  sasl: {
    mechanism: 'plain', // scram-sha-256 or scram-sha-512
    username: 'my-username',
    password: 'my-password'
  },
})

OAUTHBEARER示例

new Kafka({
  clientId: 'my-app',
  brokers: ['kafka1:9092', 'kafka2:9092'],
  // authenticationTimeout: 10000,
  // reauthenticationThreshold: 10000,
  ssl: true,
  sasl: {
    mechanism: 'oauthbearer',
    oauthBearerProvider: async () => {
      // Use an unsecured token...
      const token = jwt.sign({ sub: 'test' }, 'abc', { algorithm: 'none' })

      // ...or, more realistically, grab the token from some OAuth endpoint

      return {
        value: token
      }
    }
  },
})

该sasl对象必须包含一个名为 oauthBearerProvider 属性，一个用于返回 OAuth承载令牌的异步函数。

OAuth承载令牌必须是一个具有属性值和（可选）扩展的对象，将在SASL、OAUTHBEARER请求期间发送。

oauthbearerProvider的实现必须确保令牌在适当的时候被重用和刷线。

import { AccessToken, ClientCredentials } from 'simple-oauth2'
interface OauthBearerProviderOptions {
  clientId: string;
  clientSecret: string;
  host: string;
  path: string;
  refreshThresholdMs: number;
}

const oauthBearerProvider = (options: OauthBearerProviderOptions) => {
  const client = new ClientCredentials({
    client: {
      id: options.clientId,
      secret: options.clientSecret
    },
    auth: {
      tokenHost: options.host,
      tokenPath: options.path
    }
  });

  let tokenPromise: Promise<string>;
  let accessToken: AccessToken;

  async function refreshToken() {
    try {
      if (accessToken == null) {
        accessToken = await client.getToken({})
      }

      if (accessToken.expired(options.refreshThresholdMs / 1000)) {
        accessToken = await accessToken.refresh()
      }

      const nextRefresh = accessToken.token.expires_in * 1000 - options.refreshThresholdMs;
      setTimeout(() => {
        tokenPromise = refreshToken()
      }, nextRefresh);

      return accessToken.token.access_token;
    } catch (error) {
      accessToken = null;
      throw error;
    }
  }

  tokenPromise = refreshToken();

  return async function () {
    return {
      value: await tokenPromise
    }
  }
};

const kafka = new Kafka({
  // ... other required options
  sasl: {
    mechanism: 'oauthbearer',
    oauthBearerProvider: oauthBearerProvider({
      clientId: 'oauth-client-id',
      clientSecret: 'oauth-client-secret',
      host: 'https://my-oauth-server.com',
      path: '/oauth/token',
      // Refresh the token 15 seconds before it expires
      refreshThreshold: 15000,
    }),
  },
})

AWS IAM

省略跳过。

使用加密协议

强烈建议您在使用 SSL 当使用 PLAIN时，否则密码将会以明文传输。

自定义身份验证机制

KafkaJS支持自定义，身份验证机制,基本用不到，用到是在详细看看也不晚，知道这回事就行了。

https://kafka.js.org/docs/custom-authentication-mechanism

连接超时

等待连接成功的时间 毫秒，默认值为 1000。

new Kafka({
  clientId: 'my-app',
  brokers: ['kafka1:9092', 'kafka2:9092'],
  connectionTimeout: 3000
})

KafkaJS-请求超时

等待请求成功的时间（毫秒），默认值为 30000。

new Kafka({
  clientId: 'my-app',
  brokers: ['kafka1:9092', 'kafka2:9092'],
  requestTimeout: 25000
})

enforceRequestTimeout 可以通过设置为来禁用请求超时false。

new Kafka({
  clientId: 'my-app',
  brokers: ['kafka1:9092', 'kafka2:9092'],
  enforceRequestTimeout: false
})

KafkaJS-默认重试

retry选项可用于设置重试机制的配置，用于重试与Kafka的连接和API调用（使用生产者或消费者时）。

重试机制使用指数增长的随机化函数。https://kafka.js.org/docs/retry-detailed

如果超过最大重试次数，重试器将抛出异常 KafkaJSNumberOfRetriesExceeded 并中断。生产者将会错误冒泡到用户代码中，消费者将等待异常附带的重试时间（该时间基于尝试次数），然后执行完全重启。

new Kafka({
  clientId: 'my-app',
  brokers: ['kafka1:9092', 'kafka2:9092'],
  retry: {
    initialRetryTime: 100,
    retries: 8
  }
})

restartOnFailure

消费者用尽所有重试次数后，将调用一个异步函数，用于决定是否重启消费者（本质上是重置 consumer.run）。例如，如果需要，可以在崩溃前干净地关闭资源。该函数将接收错误信息，以便根据错误类型决定是退出应用程序还是允许其重启。

该函数具有一下签名：(error: Error) => Promise<boolean>

• 如果Promise resolves to true: consumer将会restart
• 如果Promise resolves to false: consumer将不会restart
• 如果Promise rejects: consumer将会重启
• 如果没有 restartOnFailure: consumer将会重启

请注意，该函数仅在 KafkaJS 认为可重试的错误发生时才会被调用。对于不可重试的错误，消费者不会重启，该restartOnFailure函数也不会被调用。请参阅此列表 （https://kafka.apache.org/protocol#protocol_error_codes） 以了解 Kafka 协议中的可重试错误，但请注意，一些其他错误在 KafkaJS 中仍被视为可重试，例如网络连接错误。

KafkaJS-日志记录

KafkaJS内置一个 STDOUT 输出 JSON格式的日志器，它还接受自定义日志创建器，方便您集成自己喜欢的日志库，共有 5 种 日志级别可供选择：NOTHING、ERROR、WARN、INFO、DEBUG，INFO为默认配置。

日志级别

const { Kafka, logLevel } = require('kafkajs')

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['kafka1:9092', 'kafka2:9092'],
  logLevel: logLevel.ERROR
})

要在实例化后覆盖日志级别，请setLogLevel调用单个记录器。

const { Kafka, logLevel } = require('kafkajs')

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['kafka1:9092', 'kafka2:9092'],
  logLevel: logLevel.ERROR
})
kafka.logger().setLogLevel(logLevel.WARN)

const producer = kafka.producer(...)
producer.logger().setLogLevel(logLevel.INFO)

const consumer = kafka.consumer(...)
consumer.logger().setLogLevel(logLevel.DEBUG)

const admin = kafka.admin(...)
admin.logger().setLogLevel(logLevel.NOTHING)

环境变量KAFKAJS_LOG_LEVEL也可以使用，并且它优先于代码中的配置，例如：

KAFKAJS_LOG_LEVEL=info node code.js

KafkaJS-自定义套接字工厂

为了允许自定义套接字配置，客户端接受一个可选 socketFactory 属性，该属性将用于构造任何套接字。

socketFactory 应该是一个返回与之兼容的对象的函数 net.Socket

const { Kafka } = require('kafkajs')

// Example socket factory setting a custom TTL
const net = require('net')
const tls = require('tls')

const myCustomSocketFactory = ({ host, port, ssl, onConnect }) => {
  const socket = ssl
    ? tls.connect(
        Object.assign({ host, port }, ssl),
        onConnect
      )
    : net.connect(
        { host, port },
        onConnect
      )

  socket.setKeepAlive(true, 30000)

  return socket
}

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['kafka1:9092', 'kafka2:9092'],
  socketFactory: myCustomSocketFactory,
})

代理支持

可以使用自定义套接字工厂实现对代理流量的支持，在下面的实例中，我们使用 proxy-chain 与代理服务器集成，但只要实现了套接字工场接口，就可以使用任何其他代理库。

const tls = require('tls')
const net = require('net')
const { createTunnel, closeTunnel } = require('proxy-chain')

const socketFactory = ({ host, port, ssl, onConnect }) => {
  const socket = ssl ? new tls.TLSSocket() : new net.Socket()

  createTunnel(process.env.HTTP_PROXY, `${host}:${port}`)
    .then((tunnelAddress) => {
      const [tunnelHost, tunnelPort] = tunnelAddress.split(':')
      socket.setKeepAlive(true, 60000)
      socket.connect(
        Object.assign({ host: tunnelHost, port: tunnelPort, servername: host }, ssl),
        onConnect
      )

      socket.on('close', () => {
        closeTunnel(tunnelServer, true)
      })
    })
    .catch(error => socket.emit('error', error))

  return socket
}

KafkaJS-Producing Messages

生成消息

要将消息发布到Kafka,您必须创建一个生产者，只需调用 producer 客户端的函数即可创建它：

const producer = kafka.producer();

或带有选项

const producer = kafka.producer({
  allowAutoTopicCreation: false,
  transactionTimeout: 30000
});

KafkaJS-生成消息

该方法send用于将消息发布到Kafka集群。

// 创建一个生产者向主题生成消息
const producer_run = async () => {
    const producer = kafka.producer();
    await producer.connect();
    const send_result = await producer.send({
        topic: 'test-topic',
        messages: [
            { key: 'key1', value: `key1 Hello KafkaJS user! ${Date.now()}` },
            { key: 'key2', value: `key2 Hello KafkaJS user! ${Date.now()}` },
        ],
    });

    console.log('send result:', send_result);
    if (send_result && send_result.length > 0) {
        console.log(`消息发送成功，主题: ${send_result[0].topicName}, 分区: ${send_result[0].partition}, 偏移量: ${send_result[0].baseOffset}`);
    } else {
        console.log('消息发送失败');
    }

    await producer.disconnect();
};

具有自定义分区的示例：

const create_topic = async () => {
    const admin = kafka.admin();
    await admin.connect();

    let topics = await admin.listTopics();

    console.log("已有的Topic: ", topics);

    if (topics.includes('test-topic'))
        await admin.deleteTopics({ topics: ['test-topic'] });

    topics = await admin.listTopics();
    console.log("删除test-topic后，已有的Topic: ", topics);

    if (!topics.includes('test-topic')) {
        console.log("创建Topic: test-topic");
        const topicConfig = {
            topics: [
                {
                    topic: 'test-topic',//主题名
                    numPartitions: 2,//分区数
                    replicationFactor: 1//副本数
                }],
            waitForLeaders: true//等待leader分配完成
        };

        const result = await admin.createTopics(topicConfig);
        console.log('topic created:', result);
    }

    console.log("已有的Topic: ", await admin.listTopics());

    await admin.disconnect();
};

// 创建一个生产者向主题生成消息
const producer_run = async () => {
    const producer = kafka.producer();
    await producer.connect();
    const send_result = await producer.send({
        topic: 'test-topic',
        messages: [
            { key: 'key1', value: `key1 Hello KafkaJS user! ${Date.now()}`, partition: 0 },
            { key: 'key2', value: `key2 Hello KafkaJS user! ${Date.now()}`, partition: 1 },
        ],
    });

    console.log('send result:', send_result);
    if (send_result && send_result.length > 0) {
        console.log(`消息发送成功，主题: ${send_result[0].topicName}, 分区: ${send_result[0].partition}, 偏移量: ${send_result[0].baseOffset}`);
    } else {
        console.log('消息发送失败');
    }

    await producer.disconnect();
};

// 创建一个消费者从主题消费消息
const consumer_run = async () => {

    let consumeCounter = 0;

    const consumer = kafka.consumer({ groupId: 'test-group' });
    await consumer.connect();
    await consumer.subscribe({ topic: 'test-topic', fromBeginning: true });

    await consumer.run({
        eachMessage: async ({ topic, partition, message }) => {
            consumeCounter++;
            console.log(`消费消息: consumeCounter[${consumeCounter}]`, {
                value: message.value.toString(),
                topic,
                partition
            })
        },
    });

    console.log("消费者已启动，等待消息...");
};

send方法具有以下签名：

await producer.send({
  topic: <String>,
  messages: <Message[]>,
  acks: <Number>,
  timeout: <Number>,
  compression: <CompressionTypes>,
})

KafkaJS-消息结构

单个Message结构

Key

消息 key 用于决定将消息发送到哪个分区，这对于确保与同一聚合相关的消息按顺序处理非常重要，例如，如果使用 orderId 作为键，则可以确保与该订单相关的所有消息都按顺序处理。

默认情况下，生产者配置为使用以下逻辑分发消息：

• 如果消息中指定了分区，则使用它
• 如果没有指定分区但存在密钥，则根据密钥的哈希值（murmur2）选择分区
• 如果没有分区或建，则以循环方式选择一个分区

Timestamp

每条消息都有一个时间戳，该事件戳采用UTC时间戳格式，精度为毫秒，是一个字符串，如果未提供时间戳，生产者将使用当前时间作为时间戳。消息被消费时，代理可能会根据主题配置覆盖此时间戳：

• 如果主题配置为使用 CreateTime,则将使用来自生产者消息的时间戳
• 如果主题配置为使用 LogAppendTime，则当Broker将消息附加到其Log时，时间戳将被代理本地时间覆盖。

Headers

Kafka v0.11 引入了记录头，允许您的消息携带额外的元数据，要随消息一起发送记录头，请 headers 在值中包含键。

await producer.send({
    topic: 'topic-name',
    messages: [{
        key: 'key1',
        value: 'hello world',
        headers: {
            'correlation-id': '2bfb68bb-893a-423b-a7fa-7b568cad5b67',
            'system-id': 'my-system',
        }
    }]
})

header value 可以是 字符串或字符串数组。

KafkaJS-生产到多个Topic

要同时生产到多个Topic，请使用 sendBatch。例如，在两个Topic之间迁移时，此功能非常有用。

const topicMessages = [
  {
    topic: 'topic-a',
    messages: [{ key: 'key', value: 'hello topic-a' }],
  },
  {
    topic: 'topic-b',
    messages: [{ key: 'key', value: 'hello topic-b' }],
  },
  {
    topic: 'topic-c',
    messages: [
      {
        key: 'key',
        value: 'hello topic-c',
        headers: {
          'correlation-id': '2bfb68bb-893a-423b-a7fa-7b568cad5b67',
        },
      }
    ],
  }
]
await producer.sendBatch({ topicMessages })

sendBatch 具有与send相同的函数签名，但 topic 和 messages 被替换为 topicMessages:

await producer.sendBatch({
    topicMessages: <TopicMessages[]>,
    acks: <Number>,
    timeout: <Number>,
    compression: <CompressionTypes>,
})

KafkaJS-自定义分区器

可以为生产者分配一个自定义的分区器，分区器是一个函数，它返回另一个负责分区选择的函数，如下所示：

const MyPartitioner = () => {
    return ({ topic, partitionMetadata, message }) => {
        // select a partition based on some logic
        // return the partition number
        return 0
    }
}

partitionMetadata 是具有以下结构的分区数组：

{ partitionId: <NodeId>, leader: <NodeId> }

例子：

[
    { partitionId: 1, leader: 1 },
    { partitionId: 2, leader: 2 },
    { partitionId: 0, leader: 0 }
]

createPartitioner 要使用自定义分区器，请在创建 producer 时使用该选项。

kafka.producer({ createPartitioner: MyPartitioner })

KafkaJS-默认分区器

KafkaJS 附带2个分区器：DefaultPartitioner 和 LegacyPartitioner

应该与Java Kafka客户端自带的默认分区器兼容，这对于满足连接多个Topic时共同分区的需求 DefaultPartitioner 非常重要，不然 KafkaJS和Java同时用，可想而知完蛋了。

KafkaJS-重试Retry

该选项 retry 可用于为生产者定义配置。在客户端配置 Client Configuration 部分我们了解过了。

KafkaJS-压缩

由于kafkaJS的目标是占用空间尽可能小，依赖性尽可能少，因此只有GZIP编码器是核心功能的一部分，其他编解码器可作为包使用。

GZIP

const { CompressionTypes } = require('kafkajs')

async () => {
  await producer.send({
    topic: 'topic-name',
    compression: CompressionTypes.GZIP,
    messages: [
        { key: 'key1', value: 'hello world' },
        { key: 'key2', value: 'hey hey!' }
    ],
  })
}

消费者知道如何解压缩GZIP，因此无需进一步的工作。

Snappy

Snappy 使用包 kafkajs-snappy 可以被支持。https://github.com/tulios/kafkajs-snappy

npm install --save kafkajs-snappy
# yarn add kafkajs-snappy

const {  CompressionTypes, CompressionCodecs } = require('kafkajs')
const SnappyCodec = require('kafkajs-snappy')

CompressionCodecs[CompressionTypes.Snappy] = SnappyCodec

LZ4

LZ4 使用包 kafkajs-lz4 可以被支持。https://github.com/indix/kafkajs-lz4

npm install --save kafkajs-lz4
# yarn add kafkajs-lz4

const { CompressionTypes, CompressionCodecs } = require('kafkajs')
const LZ4 = require('kafkajs-lz4')

CompressionCodecs[CompressionTypes.LZ4] = new LZ4().codec

ZSTD

Zstandard 使用包 @kafkajs/zstd。https://github.com/kafkajs/zstd

npm install --save @kafkajs/zstd
# yarn add @kafkajs/zstd

const { CompressionTypes, CompressionCodecs } = require('kafkajs')
const ZstdCodec = require('@kafkajs/zstd')

CompressionCodecs[CompressionTypes.ZSTD] = ZstdCodec()

OTHER

任何其他编解码器都可以使用现有库轻松实现。

编解码器是一个具有两个 async 函数的对象，compress 和 decompress。导入库并定义编解码器对象：

const MyCustomSnappyCodec = {
  async compress(encoder){
    return someCompressFunction(encoder.buffer);
  },
  async decompress(buffer){
    return someDecompressFunction(buffer);
  }
}

现在我们有了编解码器对象，我们可以将其包装在一个函数中并将其添加到实现中：

const { CompressionTypes, CompressionCodecs } = require('kafkajs')
CompressionCodecs[CompressionTypes.Snappy] = () => MyCustomSnappyCodec

新的编解码器现在可以与方法一起使用send，例如：

await producer.send({
    topic: 'topic-name',
    compression: CompressionTypes.Snappy,
    messages: [
        { key: 'key1', value: 'hello world' },
        { key: 'key2', value: 'hey hey!' }
    ],
})

KafkaJS-Transactions

KafkaJS提供了一个简单的接口来支持Kafka事务（事务需要Kafka版本>=0.11）

KafkaJS-在事务内发送消息

通过异步调用来初始化事务 producer.transaction()。返回的事务对象包含方法 send 和 sendBatch 其签名与生产者相同。完成后调用 transaction.commit() 或 transaction.abort() 来结束事务。具有事务感知的消费者只会读取已提交的消息。

Kafka要求事务生产者具有以下配置来保证 Exactly-once-semantics

“in-flight requests”（飞行中请求）指的是生产者（Producer）向 Kafka Broker 发送的请求，但尚未收到 Broker 确认（acknowledgement）的那些请求。 简单来说，就是“已发出但未确认”的消息批次或请求，这些请求正“在飞行中”（in flight），等待 Broker 处理并返回结果。

• max.in.flight.requests.per.connection = 1：限制每个连接的飞行中请求（in-flight requests）最多 1 个。这防止消息乱序（out-of-order），因为如果允许多个请求并行，Broker 故障时可能导致重试时顺序颠倒，破坏原子性。
• acks = all (-1)：等待所有 ISR（In-Sync Replicas，在同步副本）确认写入。这确保耐久性（durability），消息不会在部分副本丢失后被视为“成功”。
• retries = Integer.MAX_VALUE（或无限重试）：启用无限重试以处理瞬时故障（如网络抖动），结合 PID（Producer ID）和序列号机制，确保重试不会产生重复。

额外前提：

• 必须启用 enable.idempotence = true（幂等性）。
• 设置唯一的 transactional.id（事务 ID），用于协调和恢复。
• Broker 侧需配置 transaction.state.log.replication.factor >= 1 和 transaction.state.log.min.isr >= 1（单节点可设为 1）。

如果缺少这些，Kafka 只能保证 At-Least-Once 或 At-Most-Once，而非 EoS。

// 创建一个生产者向主题生成消息
const producer_run = async () => {
    const producer = kafka.producer({
        transactionalId: 'my-transactional-producer', // 为producer指定一个唯一的事务ID
        maxInFlightRequests: 1, // 只能有一个飞行请求
        idempotent: true // 开启幂等
    });
    await producer.connect();

    const transaction = await producer.transaction();

    try {
        await transaction.send({
            topic: 'test-topic',
            messages: [
                { key: 'key1', value: `key1 Hello KafkaJS user! ${Date.now()}`, partition: 0 },
                { key: 'key2', value: `key2 Hello KafkaJS user! ${Date.now()}`, partition: 1 },
            ],
        });

        // throw new Error("模拟错误，测试事务回滚");

        await transaction.commit();
        console.log('Transaction committed successfully');
    } catch (err) {
        await producer.disconnect();
        console.error('Error in producer:', err);
    }

    await producer.disconnect();
};

KafkaJS-选择一个transactionalId

transactionalId 在 Kafka 事务中不仅是唯一标识，更是“围栏”（fencing）机制的核心，用于隔离旧的“僵尸”生产者实例（zombie instances）。这确保了 Exactly-Once Semantics (EoS) 在流处理（如 read-process-write 管道）中的可靠性，尤其在分布式应用中，生产者可能因故障重启或多副本竞争。

为什么 transactionalId 需要这样设计

• 围栏机制（Fencing）：Kafka 使用 transactionalId 注册生产者到事务协调器（Transaction Coordinator）。当一个新实例（使用相同 ID）注册时，旧实例的 Producer ID (PID) 被“围栏”掉——任何来自旧实例的写入都会被 Broker 拒绝（返回 INVALID_PRODUCER_EPOCH 错误）。这防止了“僵尸”生产者（e.g., 延迟重启的进程）覆盖新数据，确保只有最新注册的生产者能写入。
• EoS 在流处理中的作用：在 Kafka Streams 或自定义管道中，读-处理-写周期必须原子。如果输入主题的分区（如 input-topic-0）的生产者 ID 不一致，重试可能导致重复处理（e.g., 订单重复扣款）。固定 transactionalId 绑定到特定分区，确保整个周期的幂等性和顺序。

如果不正确选择 ID：

• 风险：事务挂起、重复消息或 InvalidTxnStateException。
• 默认行为：Kafka 不强制唯一性，但重复 ID 会触发围栏，导致旧实例失败。

推荐选择方案：编码主题 + 分区

您提到的简单方案 "myapp-producer-" + topic + "-" + partition 是最佳实践，尤其适合分区级流处理：

格式示例：

• 对于主题 user-events 的分区 0："myapp-producer-user-events-0"
• 对于主题 orders 的分区 3："myapp-producer-orders-3"

优点：

• 唯一性：自动避免跨分区冲突（每个分区独立事务）。
• 可扩展：易于监控（日志中可见分区绑定）。
• 恢复友好：重启时，使用相同 ID 恢复挂起事务。

高级变体（多输入主题）：

如果管道涉及多个输入主题，用组合："myapp-producer-" + inputTopic + "-" + partition + "-" + instanceId（instanceId 如主机名或线程 ID，防多实例冲突）。 对于 Streams 应用，Kafka Streams 内部自动生成基于 application.id + 分区，但自定义生产者需手动编码。

// 为每个分区创建独立生产者（或动态 ID）
const getTransactionalId = (topic, partition) => `myapp-producer-${topic}-${partition}`;

const producer = kafka.producer({
  transactionalId: getTransactionalId('test-topic', 0),  // 固定分区 ID
  idempotent: true,
  // ... 其他 EoS 配置
});

多分区：用线程池或 Streams 库，每个分区一个生产者实例。

https://www.confluent.io/blog/transactions-apache-kafka/

KafkaJS-发送偏移量

要将偏移量作为事务的一部分发送，即只有事务成功时才会提交偏移量，请使用 transaction.sendOffsets() 方法。当我们希望事务在“消费-转换-生产”循环中生成来自消费者的消息时，这都是必需的

await transaction.sendOffsets({
  consumerGroupId, topics
})

topics具有以下结构：

[{
  topic: <String>,
  partitions: [{
    partition: <Number>,
    offset: <String>
  }]
}]

KafkaJS-Topic Partition ConsumerGroup Offset关系

Kafka中消费者组（groupId）、主题（topic）、分区（partition）三者之间的关系，是Kafka高可扩展性和高可靠性的核心机制之一。

KafkaJS-三者的核心关系

一个主题Topic 可以有多个分区（Partition）

• 每个分区只会被同一个消费者组中的一个消费者实例消费
• 不同消费者组之间互不影响

Topic: orders
Partitions: 0, 1, 2

KafkaJS-消费者组（Group）决定消费 并行度

假设我们有：主题 orders 分区数 3 消费者组ID group-A

即：一个分区同组内能有一个消费者读取，但多个消费者组可以读取同一个分区

KafkaJS-同一个groupId内消费者之间的分区分配Rebalance

当组内的消费者数量变化时（例如新消费者加入、旧消费者崩溃），Kafka会进行Rebalance再平衡：

• Kafka协调者（Group Coordinator）会重新分配分区
• 每个分区仍然只会被组内一个消费者读取
• 但消费者数量可以多于或少于分区数

KafkaJS-不同groupId各自独立消费

即使是同一个Topic,不同的groupId会各自维护offset消费进度，它们不会互相影响。

举例：

• group-A 消费 orders 到 offset 200
• group-B 也消费 orders，它的offset可能还在50,两者互不干扰

这就是Kafka支持一条消息可被多个独立系统同时消费的原因（日志分析系统、警告系统、数据仓库等可共享同一Topic）。

可视化示意图

Topic: orders
├── Partition 0 ───► consumer-1 (group-A)
├── Partition 1 ───► consumer-2 (group-A)
├── Partition 2 ───► consumer-3 (group-A)

Topic: orders
├── Partition 0 ───► consumer-X (group-B)
├── Partition 1 ───► consumer-Y (group-B)
├── Partition 2 ───► consumer-Z (group-B)

group-A 和 group-B 都在消费同一个Topic，但各自维护自己的消费进度。

KafkaJS-Offset偏移量与groupId的关系

Kafka为每个消费者组维护消费进度 Offset

__consumer_offsets
   ├── key: (groupId, topic, partition)
   └── value: offset

举例

(groupId=group-A, topic=orders, partition=1) → offset=1056

所以：

• groupId决定了offset存储空间
• 如果你改了groupId,Kafka会认为是新的消费者组，从头（或最新位置）开始消费

KafkaJS-Consuming Messages

消费者组允许一组机器或进程协调对Topic列表的访问，从而在消费者之间分配负载，当一个消费者发生故障时，负载会自动分配给组中的其他成员。从Kafka Broker的角度来看， 消费者组在集群内必须具有唯一的组ID。

创建消费者

const consumer = kafka.consumer({ groupId: 'test-group' });

订阅一些主题

await consumer.connect();
await consumer.subscribe({ topics: ['topic-A'] })

// You can subscribe to multiple topics at once
await consumer.subscribe({ topics: ['topic-B', 'topic-C'] })

// It's possible to start from the beginning of the topic
await consumer.subscribe({ topics: ['topic-D'], fromBeginning: true })

或者，您可以订阅任何与正则表达式匹配的Topic

await consumer.connect()
await consumer.subscribe({ topics: [/topic-(eu|us)-.*/i] })

提供正则表达式时，消费者将不会匹配订阅后创建的主题。如果您的代理有 topic-A 和 topic-B，并且您订阅了 /topic-.*，那么 topic-C 创建后，您的消费者将不会自动订阅 topic-C。

KafkaJS 为您提供了两种处理数据的方法 eachMessage 和 eachBatch。

KafkaJS-eachMessage

该 eachMessage 处理程序提供了一个便捷易用的API，可以一次向您的函数提供一条消息。它基于 eachBatch 实现，并会按照您配置的间隔自动提交偏移量和心跳。如果您刚开始使用Kafka消费者，那么这是一个不错的起点。

// 创建一个消费者从主题消费消息
const consumer_run = async () => {

    let consumeCounter = 0;

    const consumer = kafka.consumer({ groupId: 'test-group', sessionTimeout: 30000 });
    await consumer.connect();
    await consumer.subscribe({ topic: 'test-topic', fromBeginning: true });

    await consumer.run({
        eachMessage: async ({ topic, partition, message, heartbeat, pause }) => {
            consumeCounter++;
            console.log(`消费消息: consumeCounter[${consumeCounter}]`, {
                key: message.key.toString(),
                value: message.value.toString(),
                topic,
                partition,
                headers: message.headers,
            })
        },
    });

    console.log("消费者已启动，等待消息...");
};

请注意，eachMessage 处理程序的阻塞时间不应超过配置的 session timeout,否则消费者将被从组中移除。如果您的工作负载导致单个消息的处理时间非常慢，那么您应该增加会话超时时间，或者定期使用 heartbeat 处理程序负载中公开的函数。该 pause 函数提供了方便的调用 consumer.pause({topic, partitions:[partition]})。它将暂停当前Topic Partition，并返回一个允许您稍后恢复消费的函数。

KafkaJS-eachBatch

有些用例需要直接处理批次。此处理程序将为您的函数提供批次，并提供一些实用函数，以提高您的代码灵活性：resolveOffset、 hearbeat、commitOffsetsIfNecessary、uncommittedOffsets、isRunning、isStale、pause。所有已解析的偏移量将在函数执行后自动提交。

与使用 eachBatch 相比，直接使用 eachMessage 是被认为更高级的用例，因此您必须了解会话超时和心跳是如何连接的。

// 创建一个消费者从主题消费消息
const consumer_run = async () => {
    let consumeCounter = 0;

    const consumer = kafka.consumer({ groupId: 'test-group' });
    await consumer.connect();
    await consumer.subscribe({ topic: 'test-topic', fromBeginning: true });

    await consumer.run({
        eachBatchAutoResolve: true,
        eachBatch: async ({
            batch,
            resolveOffset,
            heartbeat,
            commitOffsetsIfNecessary,
            uncommittedOffsets,
            isRunning,
            isStale,
            pause
        }) => {
            for (let message of batch.messages) {
                consumeCounter++;
                console.log(`消费消息: consumeCounter[${consumeCounter}]`, {
                    topic: batch.topic,
                    partition: batch.partition,
                    highWatermark: batch.highWatermark,
                    message: {
                        offset: message.offset,
                        key: message.key.toString(),
                        value: message.value.toString(),
                        headers: message.headers,
                    }
                });

                // 标记已消费
                resolveOffset(message.offset);
                // 心跳
                await heartbeat();
            }
        }
    });

    console.log("消费者已启动，等待消息...");
};

• eachBatchAutoResolve：配置了批处理自动解析。如果设置为true，KafkaJS将在eachBatch未抛出错误时自动提交批次的最后一个偏移量，默认值为 true。
• batch.highWatermark：是主题分区内最后一个已提交的偏移量，它可用于计算滞后（lag）。
• resolveOffset()：用于标记批次中的消息为已处理。在错误情况下，消费者将自动提交已解析的偏移量。
• heartbeat()：Promise<void> 可用于根据消费者配置中设置的 heartbeatInterval 值向Broker发送心跳，这意味着如果您在 heartbeatInterval 之前调用 heartbeat()，它将被忽略。
• commitOffsetsNecessary(offsets?)：Promise<void> 用于基于 autoCommit 配置（autoCommitInterval）和（autoCommitThreshold）提交偏移量。请注意，如果未调用 commitOffsetsIfNecessary，则autocommit不会在eachBatch中发生而是在eachBatch调用后。
• uncommittedOffsets()：返回所有尚未提交的按主题 分区组织的偏移量
• isRunning()：返回消费者是否处于运行状态，如果是则返回true、否则返回false；
• isStale()：返回批次中的消息是否通过其他操作被渲染为过时并应被丢弃。例如，当调用 consumer.seek 时，批次中的消息应被丢弃，因为它们不在我们seek的偏移量处。
• pause()：可用于暂停当前主题 分区的消费者，在此点之前解析的所有偏移量将被提交（受 eachBatchAutoResolve 和 autoCommit 影响）。在批次中间抛出错误以暂停而不解析当前偏移量。或者，禁用 eachBatchAutoResolve。返回的函数可用于恢复主题-分区的处理。

KafkaJS-分区感知并发

默认情况下，eachMessage对每个分区中的每条消息按顺序调用，为了同时处理多条消息，可以增加 partitionConsumedConcurrently 选项

// 创建一个消费者从主题消费消息
const consumer_run = async () => {
    let consumeCounter = 0;

    const consumer = kafka.consumer({ groupId: 'test-group' });
    await consumer.connect();
    await consumer.subscribe({ topic: 'test-topic', fromBeginning: true });

    consumer.run({
        partitionsConsumedConcurrently: 2,
        eachMessage: async ({ topic, partition, message }) => {
            consumeCounter++;
            console.log(`消费消息: consumeCounter[${consumeCounter}]`, {
                topic,
                partition,
                message: {
                    offset: message.offset,
                    key: message.key.toString(),
                    value: message.value.toString(),
                    headers: message.headers,
                }
            });
        }
    });

    console.log("消费者已启动，等待消息...");
};

同一分区中的消息仍保证按顺序处理，但来自多个分区的消息可以同时处理。如果消息 eachMessage 由异步工作（例如网络请求或其他I/O）组成，这可以提高性能。如果 eachMessage 消息完全同步，则不会有任何影响。

如果您使用 eachBatch 情况也是如此，给定 partitionsConsumedConcurrently > 1，将能够同时处理多个批次。

设置 partitionsConsumedConcurrently 的指导原则是，该值不应大于实际使用的分区数量，根据您的工作负载是否受CPU限制，将其设置为高于逻辑CPU核心数可能也无益，建议从较低的值开始，然后测量增加该值是否能提高吞吐量。

KafkaJS-自动提交

消息总是以批次形式从Kafka获取，即使使用eachMessage处理程序。所有解析的偏移量将从处理整个批次后提交到Kafka。

在批次期间定期提交偏移量允许消费者从组再平衡、过时元数据和其他问题中恢复，而无需完成整个批次。然而，更频繁的提交会增加网络流量并减慢处理速度。Auto-commit提供了更多灵活性来提交偏移量；有两种可用模式；

autoCommitInterval：消费者将在给定时间段后提交偏移量，例如，2秒，值以毫秒为单位，默认值 null

consumer.run({
  autoCommitInterval: 5000,
  // ...
})

autoCommitThreshold：消费者将在解析给定数量的消息后提交偏移量，例如，一百条消息。默认值 null

consumer.run({
  autoCommitThreshold: 100,
  // ...
})

同时使用两种模式也是可能的，消费者将在任何一种用例（间隔或消息数量）发生时提交偏移量。

autoCommit：高级选项，用于完全禁用自动提交，相反，您可以手动提交偏移量，默认值：true。

KafkaJS-手动提交

当禁用 autoCommit 时，您仍然可以手动提交消息偏移量，有几种不同的方式；