工具使用集|Kafka：官方文档简明解读

Kafka 上

前言：

Active Mq 似乎已经被Kafak替代了？🤔，确实现在项目似乎大多数都是由了Kafka这样的消息日志存储中心。那么我们就来学习一下kafka的知识吧。长话短说，我们开始认识一下Kafka吧！

介绍

kafka的核心：

​ 事件流（event stream），它可用于一些实时数据的处理，保证数据的有效性。它是可以在各种行业中构建实时数据处理和分发系统的关键工具。

kafka的运作简单来说分为两种：

通过高性能 TCP 网络协议进行通信的服务器和客户端组成

• Servers

kafka运行在集群上面，而kafka运行的服务器形成存储层又被称为代理（brokers）。

• Clients

kafka的客户端，可以通过编写相应的kafka程序来对 servers 存储的数据进行读取、写入、处理流。

概念术语：

在kafka中

​ 读取和写入数据又被称为记录或者消息

​ 事件（Event）通常是称为消息或记录的数据单元。事件是一种以结构化或非结构化形式记录的数据，用于表示某个特定时间点发生的事情或包含有关某个事件的信息

​ 事件又包括：

​ 关键字、值、时间戳和元数据头

​ 生产者（Producers）：是将事件发布（写入）到kafka中的客户端程序

​ 消费者（Consumers ）：是订阅（读取和处理）kafka中存储数据的客户端程序

​ 生存者和消费者两者直接完全解耦。生产者永远不需要等待消费者。

​ 主题(Topics)：上面中提到的事件就是存储在主题中的。主题可以看作是文件系统的文件夹，事件是文件。

​ 主题可以有多个生产者写入事件、也可以有多个消费者订阅事件。事件被消费了并不会被删除。

​ 主题是分区，就是说主题分布在不同的 kafka 代理（broker 还记得broker的概念吗？ ）上。（这样说吧，如果kafka部署有两个broker，那么此时我们创建了一个 topics ,那么在这两个broker 中都会存在这个 topics ）。Kafka保证了对于给定的主题-分区，任何消费者总是按照事件写入的顺序来读取该分区的事件。

注：

producer client 1 ：发送事件，会将事件发送到P1和P3分区（这些分区可以分布在一个或多个Broker上，具体取决于Kafka集群的设置。）。

producer client 2 ：发送事件到P3、P4.发送会根据事件顺序进入，那么相应的读取也是会是相同的顺序读取

• watch talks from Kafka Summit, the main conference of the Kafka community.

Quick Start（Docker部署Kafka）

官网的kafka配置好繁琐，学习Kafka不是浪费事件在部署上面的。

我们使用docker来帮忙快速开始吧

默认大家都安装了，如果没有的话可以看看Docker系列-Linux-Centos8环境安装Docker

提醒：如果大家是用虚拟机执行的话，建议打一个快照。后续如果有操作错误的话，就执行回滚快照就好了。

上述步骤中的阿里云镜像加速，能配置最好配置，不然，下面pull kafka 镜像可能会出现问题。无法拉去。

curl -sSL https://raw.githubusercontent.com/bitnami/containers/main/bitnami/kafka/docker-compose.yml > docker-compose.yml
docker pull bitnami/kafka:latest

我们提前先把docker-compose.yml 文件下载下来。我们到时候可以直接使用这个配置来启动我们配置得kafak镜像。

2.使用docker compose进行安装

通过我们提前下载的docker compse文件进行启动

# Copyright VMware, Inc.
# SPDX-License-Identifier: APACHE-2.0

version: "2"

services:
  kafka:
    image: 'bitnami/kafka:latest'
    ports:
      - "9092:9092"
    volumes:
      - "kafka_data:/bitnami"
    environment:
      # KRaft settings
      - KAFKA_CFG_NODE_ID=0
      - KAFKA_CFG_PROCESS_ROLES=controller,broker
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093
      # Listeners
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://:9092
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=PLAINTEXT
volumes:
  kafka_data:
    driver: local

version: "2"：这是 Docker Compose 文件的版本号。

services：定义了要创建的服务及其配置。

kafka 服务：

• image: 'bitnami/kafka:latest'：指定了使用的 Kafka 镜像。这里使用了 Bitnami 提供的 Kafka 镜像，并指定使用最新版本（latest）。

• ports：指定了服务的端口映射，将主机的端口 9092 映射到容器的端口 9092。

• volumes：定义了数据卷的映射，将主机上的 kafka_data 数据卷映射到容器内的 /bitnami 目录，这可以用来存储 Kafka 数据。

•   environment
  

  ：设置了一系列环境变量，用于配置 Kafka 服务的各种参数，包括：

  • KAFKA_CFG_NODE_ID：指定 Kafka 节点的ID。

  • KAFKA_CFG_PROCESS_ROLES：定义 Kafka 进程的角色，这里包括 controller 和 broker。

  • KAFKA_CFG_CONTROLLER_QUORUM_VOTERS：指定控制器的投票者。

  • KAFKA_CFG_LISTENERS：定义 Kafka 服务的监听器。

  • KAFKA_CFG_ADVERTISED_LISTENERS：定义 Kafka 服务的广告监听器。

  • KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP：指定监听器的安全协议。

  • KAFKA_CFG_CONTROLLER_LISTENER_NAMES：定义控制器监听器的名称。

  • KAFKA_CFG_INTER_BROKER_LISTENER_NAME：定义经纪人之间的监听器名称。

volumes：定义了一个名为 kafka_data 的数据卷，用于持久化存储 Kafka 数据。

4.按照docker-compose，并且提权，查看版本是否可用

curl -L "https://github.com/docker/compose/releases/latest/download/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose

chmod +x /usr/local/bin/docker-compose

docker-compose --version

注意：如果无法下载，就是停留在curl的下载进度，那么可能需要更新一下，切记别更新（sudo yum update），不然恢复可不好，记得快照备份。不然会崩溃的！

sudo yum install openssl

3.启动docker kafka

docker-compose up -d 

注意：使用我们之前下载下来得docker-compose.yml，配置文件中的images 需要对应docker images中的tag。如果不存在对应得tag，它则会帮忙下载。不过，我们也可以提前下载。

这一步，有一个点麻烦。就是bitnami/kafka:latest 镜像不好下载。如果存在网络限制的原因。可以参考一下方式，如果镜像可以下载直接略过下面步骤

我尝试过即使配置了阿里云镜像也无法下载下来。如果，想要下载还是的需要通过代理进行下载访问。

如果，你有代理则可以进行如下配置，如果没有的话，更换其他的镜像下载即可。（不然，其他的镜像还是需要最新的kakfa。不然，你还得配置Zookeeper）

1.在 /etc/systemd/system/docker.service.d 目录下，创建一个

sudo mkdir -p /etc/systemd/system/docker.service.d
sudo touch /etc/systemd/system/docker.service.d/http-proxy.conf

2.添加完毕之后，在文件中添加代理信息

[Service]
Environment="HTTP_PROXY=http://0.0.0.0:8080/"
Environment="HTTPS_PROXY=http://0.0.0.0:8080/"
Environment="NO_PROXY=localhost,127.0.0.1"

以上内容，填写你的代理服务器。其中Environment="NO_PROXY=localhost,127.0.0.1"这个可有可无，如果你不想通过代理服务器的可以配置。

3.重启docker的守护进程以及重启

sudo systemctl daemon-reload
sudo systemctl restart docker

如果，根据以上配置，我是可以成功下载到下载量最高的kafka镜像的，并且启动和配置过程都很简单。

！干，kafka还挺麻烦的，可恶的网络啊。

也可以通过配置

export http_proxy=http://your_proxy_server:your_proxy_port
export https_proxy=http://your_proxy_server:your_proxy_port

取消

unset http_proxy
unset https_proxy

4.进入docker kafka内部

docker exec -it 40d06df75f07 /bin/bash

5.停止docker kafka

docker-compose down //这会停止和移除与 docker-compose.yml 文件中定义的服务相关的容器，并清理相关的网络和数据卷。确保在包含 docker-compose.yml 文件的目录中执行该命令。

docker-compose stop  //如果只想停止 Kafka 服务，而不清理容器和数据卷

总结一下：其实如果你能够访问外网，那么以上步骤对你来说就是粘贴复制了，如果无法访问外网，那就要头疼了。🤔

不多说，开始我真正得学习之旅了。不过，在开始学习之前，我们在完善一下kafka相关得知识吧。

kafka 配置

Kafka使用属性文件格式的键值对进行配置。这些值可以从文件或以编程方式提供。（一般是.property文件）

Broker Configs

基本配置如下:

advertised.listeners 是 Kafka 中的一个配置属性，用于告知客户端它们应该使用哪些监听器来连接到 Kafka Broker。它主要用于发布 Broker 的地址，以供客户端通信使用，这个地址可以不同于 listeners 配置属性。

以下是关于 advertised.listeners 的关键信息：

advertised.listeners 允许在复杂的网络环境中正确配置 Kafka Broker，以确保客户端能够正常连接到它们。

auto.create.topics.enable 是 Kafka 中的一个配置属性，用于控制是否允许自动创建主题（topics）。

以下是关于 auto.create.topics.enable 的关键信息：

auto.create.topics.enable 允许控制是否自动创建主题，这在某些情况下可能很有用，但在生产环境中，可能希望将其设置为 false，以确保主题的创建是明确和受控的，从而避免潜在的问题。

auto.leader.rebalance.enable 是 Kafka 中的一个配置属性，用于启用或禁用自动分配主题分区领导者的功能。以下是关于此配置属性的关键信息：

auto.leader.rebalance.enable 允许控制是否启用自动分配分区领导者并定期检查领导者的平衡情况。这有助于确保 Kafka 集群的性能和可用性，因为它可以避免领导者的不均衡分布。

background.threads 是 Kafka 中的一个配置属性，用于设置用于各种后台处理任务的线程数量。以下是有关此配置属性的关键信息：

background.threads 允许配置 Kafka 服务器用于执行各种后台处理任务的线程数量。增加线程数量可以提高 Kafka 服务器的并发性能，但也需要监控系统资源的使用情况以避免过度消耗资源。

broker.id 是 Kafka 中的一个重要配置属性，用于设置 Kafka 服务器的经纪人（broker）ID。以下是有关此配置属性的关键信息：

broker.id 是 Kafka 集群中每个经纪人的唯一标识符。在配置 Kafka 服务器时，必须为每个经纪人分配一个唯一的ID，以确保集群的正确运行。如果未手动配置，Kafka 将自动生成经纪人ID，但这可能会导致与 ZooKeeper 生成的ID 冲突，因此最好由管理员手动配置。

compression.type 是 Kafka 中的一个配置属性，用于指定给定主题（topic）的最终压缩类型。以下是有关此配置属性的关键信息：

compression.type 允许为特定主题选择消息的压缩类型，以优化存储和传输效率。可以根据需求选择不同的压缩编解码器，或者使用生产者设置的原始压缩方式。

control.plane.listener.name 是 Kafka 中的一个配置属性，用于指定控制器（controller）与代理（broker）之间通信所使用的监听器名称。以下是有关此配置属性的关键信息：

codelisteners = INTERNAL://192.1.1.8:9092, EXTERNAL://10.1.1.5:9093, CONTROLLER://192.1.1.8:9094
listener.security.protocol.map = INTERNAL:PLAINTEXT, EXTERNAL:SSL, CONTROLLER:SSL
control.plane.listener.name = CONTROLLER

control.plane.listener.name 用于配置 Kafka 代理以便控制器可以通过指定的监听器建立连接。这有助于确保代理和控制器之间的通信正常工作，并且能够选择适当的安全协议。

controller.listener.names 是 Kafka 中的一个配置属性，用于指定控制器（controller）使用的监听器名称列表。以下是有关此配置属性的关键信息：

arduinoCopy code
controller.listener.names = INTERNAL, EXTERNAL

controller.listener.names 用于配置 Kafka 控制器以指定其可以使用的监听器名称列表。这有助于确保在 KRaft 模式下运行时，控制器可以与代理建立正确的连接，以管理和控制 Kafka 集群。注意，在使用基于 ZooKeeper 的控制器时，不应设置此配置。

controller.quorum.election.backoff.max.ms 是 Kafka 中的一个配置属性，用于指定开始新选举之前的最大等待时间（以毫秒为单位）。以下是有关此配置属性的关键信息：

arduinoCopy code
controller.quorum.election.backoff.max.ms = 5000

controller.quorum.election.backoff.max.ms 用于控制在选举失败后，Kafka 控制器等待启动新选举的最大时间。这有助于维护选举的稳定性和可靠性，以确保 Kafka 集群的正常运行。

replica.lag.time.max.ms 是 Kafka 配置属性之一，用于确定在什么情况下主题的复制副本（replica）将被移除。以下是有关此配置属性的关键信息：

luaCopy code
replica.lag.time.max.ms = 60000  # 将延迟时间设置为 60 秒

replica.lag.time.max.ms 是一个用于控制 Kafka 复制副本同步行为的配置属性。通过设置此值，可以调整何时将复制副本从同步副本集中移除，以确保数据的一致性和可用性。

replica.socket.receive.buffer.bytes 是 Kafka 的一个配置属性，它用于确定用于复制数据的网络请求的领导者（leader）的套接字接收缓冲区大小。以下是有关此配置属性的详细信息：

pythonCopy code
replica.socket.receive.buffer.bytes = 131072  # 将缓冲区大小增加到 128 kibibytes

通过调整 replica.socket.receive.buffer.bytes 等网络相关的配置属性，可以根据 Kafka 集群的规模和工作负载来优化性能，以确保数据的可靠复制和高吞吐量。

replica.socket.timeout.ms 是 Kafka 的一个配置属性，用于设置网络请求的套接字超时时间。以下是有关此配置属性的详细信息：

请注意，在特定场景下，可能需要根据 Kafka 集群的工作负载和网络状况来调整此值，以确保数据的可靠复制和性能。

在特定场景下，可能需要根据 Kafka 集群的工作负载和网络状况来调整此值，以确保数据的可靠复制和性能。

request.timeout.ms 是 Kafka 客户端的一个配置属性，用于控制客户端等待请求响应的最大时间。以下是有关此配置属性的详细信息：

注:

Broker Configs 更多配置查看

Topic-Level Configs

主题相关的配置既有服务器默认值，也有可选的每个主题覆盖。如果我们主动显示得配置了，那么默认就不会生效。

cleanup.policy 是 Kafka 配置中的一个重要属性，用于指定日志段的数据保留策略。以下是有关此配置属性的详细信息：

根据的应用场景和数据保留需求，可以选择适当的 cleanup.policy 设置。如果需要确保历史消息不会无限制地增长，可以考虑使用 "delete" 策略。如果需要保留每个键的最新状态，可以选择 "compact" 策略。或者，根据需要组合这两种策略以兼顾存储和数据保留的要求。

compression.type 是 Kafka 配置中的一个属性，用于指定主题（topic）的最终压缩类型。以下是有关此配置属性的详细信息：

根据的需求，可以根据以下几个方面来选择 compression.type 的设置：

• 如果希望节省存储空间并可以容忍稍微增加的 CPU 开销，可以选择一种压缩算法（例如 "zstd"、"lz4"、"snappy" 或 "gzip"）。
• 如果消息已经在生产者端使用压缩算法进行了压缩，并且希望保留原始压缩方式，可以选择 "producer"。
• 如果希望消息以未经压缩的形式存储以提高写入和读取速度，并且不太关心存储空间，可以选择 "uncompressed"。

选择合适的压缩类型通常取决于的具体用例和对存储效率与性能的权衡需求。

delete.retention.ms 是 Kafka 配置中的一个属性，用于指定在已经进行了日志压缩（log compaction）的主题上保留删除标记（delete tombstone markers）的时间。以下是有关此配置属性的详细信息：

根据的需求，可以根据以下几个方面来选择 delete.retention.ms 的设置：

• 如果希望删除标记（删除墓碑标记）在 Kafka 主题中保留更长的时间，以确保消息删除操作被记录并传播给所有消费者，可以将此值设置为较大的值。
• 如果不需要保留删除标记，可以将此值设置为 0，以立即清除已删除的消息。

设置 delete.retention.ms 的值通常会受到应用程序需求、数据保留策略和合规性要求的影响。不同的应用程序可能需要不同的保留时间，因此应该根据具体情况进行设置。

file.delete.delay.ms 是 Kafka 配置中的一个属性，用于指定从文件系统中删除文件之前等待的时间。以下是有关此配置属性的详细信息：

根据的需求，可以根据以下几个方面来选择 file.delete.delay.ms 的设置：

• 如果希望 Kafka 在删除文件之前等待一段时间，以确保完成一些相关操作，例如文件备份、传输或其他处理，可以将此值设置为相应的延迟时间。
• 如果不需要等待，并且希望 Kafka 尽可能快地删除不再需要的文件，可以将此值设置为 0。

设置 file.delete.delay.ms 的值通常受到磁盘空间的管理、日志维护操作和数据处理流程的影响。不同的应用程序和环境可能需要不同的设置。

flush.messages 是 Kafka 配置中的一个属性，它允许指定一个间隔，在该间隔内将强制执行对写入日志的数据的 fsync 操作。以下是有关此配置属性的详细信息：

一般来说，不建议显式设置 flush.messages。相反，Kafka 倾向于依赖于操作系统的后台刷新机制，而不是在每条消息写入后都执行 fsync 操作。这是因为频繁的 fsync 操作可能会对性能产生不利影响，而 Kafka 使用复制来确保数据的持久性。

如果有特殊的需求，需要更精细地控制数据的持久性，可以考虑在特定的主题上设置不同的 flush.messages 值，以满足特定需求。但在大多数情况下，建议保持默认设置。

flush.ms 是 Kafka 配置中的一个属性，它允许指定一个时间间隔，在该间隔内将强制执行对写入日志的数据的 fsync 操作。以下是有关此配置属性的详细信息：

一般来说，不建议显式设置 flush.ms。相反，Kafka 倾向于依赖于操作系统的后台刷新机制，而不是在固定时间间隔内执行 fsync 操作。这是因为频繁的 fsync 操作可能会对性能产生不利影响，而 Kafka 使用复制来确保数据的持久性。

如果有特殊的需求，需要更精细地控制数据的持久性，可以考虑在特定的主题上设置不同的 flush.ms 值，以满足特定需求。但在大多数情况下，建议保持默认设置。

注:

Topic-Level Configs 查看更多配置

Producer Configs

生产者的配置：

key.serializer 是 Kafka 配置中的一个属性，它用于指定用于序列化消息键（message key）的序列化器类。以下是有关此配置属性的详细信息：

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

Kafka 使用 StringSerializer 来序列化消息的键。

根据的使用情况选择合适的键序列化器非常重要。如果消息键是字符串，则通常使用 StringSerializer。如果键是复杂对象，可能需要编写自定义的序列化器以确保正确序列化。请根据的数据和需求选择适当的序列化器。

value.serializer 是 Kafka 配置中的一个属性，它用于指定用于序列化消息值（message value）的序列化器类。以下是关于此配置属性的详细信息：

Copy code
value.serializer=org.apache.kafka.common.serialization.StringSerializer

根据的使用情况选择合适的值序列化器非常重要。如果消息值是字符串，则通常使用 StringSerializer。如果值是复杂对象，可能需要编写自定义的序列化器以确保正确序列化。请根据的数据和需求选择适当的序列化器。

bootstrap.servers 是 Kafka 配置中的一个关键属性，它用于指定用于建立与 Kafka 集群的初始连接的主机/端口对列表。以下是有关此配置属性的详细信息：

makefileCopy code
bootstrap.servers=host1:9092,host2:9092,host3:9092

bootstrap.servers=host1:9092,host2:9092,host3:9092

客户端在建立初始连接时使用这些服务器

bootstrap.servers 是 Kafka 客户端配置中的关键属性，它指定了与 Kafka 集群建立初始连接的服务器地址。正确配置这个属性是使用 Kafka 的关键步骤之一，以确保客户端可以与集群进行通信。

buffer.memory 是 Kafka 生产者的一个关键配置属性，用于指定生产者用于缓冲等待发送到服务器的记录的总内存字节数。以下是有关此配置属性的详细信息：

arduinoCopy code
buffer.memory=67108864

buffer.memory 用于控制 Kafka 生产者的内存缓冲区大小，确保记录可以按照适当的速率发送到 Kafka 服务器，同时避免在发送速率过快时出现内存问题。可以根据生产者的需求和可用内存来调整此配置。

compression.type 是 Kafka 生产者的一个重要配置属性，用于指定生产者生成的所有数据的压缩类型。以下是关于此配置属性的详细信息：

goCopy code
compression.type=gzip

compression.type 允许控制 Kafka 生产者是否压缩生成的消息，并可以根据需求选择合适的压缩算法。选择适当的压缩类型可以帮助在减小网络带宽占用和提高传输效率之间取得平衡。

retries 是 Kafka 生产者的一个配置属性，它用于设置在发生潜在的临时错误时是否进行消息的重试发送。以下是关于此配置属性的详细信息：

makefileCopy code
retries=3

retries 允许控制 Kafka 生产者在发生潜在的临时错误时是否进行消息的重试。通常情况下，建议将其保留为默认值，以便 Kafka 生产者自动处理重试。但是，如果需要更精确地控制重试行为，可以根据需要设置此属性。如果启用幂等性（通过 enable.idempotence 配置），则 retries 的值必须大于0。

ssl.key.password 是 Kafka SSL 配置中的一个属性，用于指定私钥的密码。以下是关于此配置属性的详细信息：

vbnetCopy code
ssl.key.password=myPrivateKeyPassword

ssl.key.password 是 Kafka SSL 配置中的一个重要属性，用于指定私钥的密码，以确保私钥的安全性和可用性。在设置 SSL 安全通信时，必须正确配置此属性。

ssl.keystore.certificate.chain 是 Kafka SSL 配置中的一个属性，用于指定 SSL 密钥库中的证书链。以下是有关此配置属性的详细信息：

cssCopy codessl.keystore.certificate.chain=-----BEGIN CERTIFICATE-----
MIIDizCCAnOgAwIBAgIUD7u3YKmG63v...
-----END CERTIFICATE-----

ssl.keystore.certificate.chain 是 Kafka SSL 配置中的一个重要属性，用于指定 SSL 密钥库中的证书链，以确保 SSL 连接的安全性和正确性。在设置 SSL 安全通信时，必须正确配置此属性。

ssl.keystore.key 是 Kafka SSL 配置的一部分，用于指定 SSL 密钥库中的私钥。以下是有关此配置属性的详细信息：

vbnetCopy codessl.keystore.key=-----BEGIN PRIVATE KEY-----
MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQDU37/y0C7XLM5A
...
-----END PRIVATE KEY-----

通常，私钥文件通常需要密码保护，因此可能还需要指定 ssl.key.password 属性，以提供私钥的密码。

ssl.keystore.key 是 Kafka SSL 配置的一个重要属性，用于指定 SSL 密钥库中的私钥，以确保 SSL 连接的安全性和正确性。在设置 SSL 安全通信时，必须正确配置此属性。

ssl.keystore.location 是 Kafka SSL 配置的一部分，用于指定 SSL 密钥库文件的位置。以下是有关此配置属性的详细信息：

javascriptCopy code
ssl.keystore.location=/path/to/your/keystore.jks

ssl.keystore.location 是 Kafka SSL 配置的一个重要属性，用于指定 SSL 密钥库文件的位置，以确保 SSL 连接的安全性和正确性。在设置 SSL 安全通信时，必须正确配置此属性。

ssl.keystore.password 是 Kafka SSL 配置的一部分，用于指定密钥库文件的密码。以下是有关此配置属性的详细信息：

Copy code
ssl.keystore.password=your_keystore_password

ssl.keystore.password 是 Kafka SSL 配置的一个重要属性，用于指定密钥库文件的密码，以确保 SSL 连接的安全性和正确性。在设置 SSL 安全通信时，必须正确配置此属性，以保护密钥库文件的访问。

注:

Broker Configs 更多配置查看

Consumer Configs

消费者配置如下：

key.deserializer 是 Kafka 生产者或消费者的配置选项之一，用于指定如何反序列化 Kafka 记录的键（key）。以下是关于 key.deserializer 的详细信息：

key.deserializer=com.example.CustomKeyDeserializer

key.deserializer 是 Kafka 消费者的一个关键配置选项，它定义了如何将 Kafka 消息的键反序列化为应用程序可以处理的格式。确保正确配置此选项对于消费者的正确运行非常重要。

value.deserializer 是 Kafka 生产者或消费者的配置选项之一，用于指定如何反序列化 Kafka 记录的值（value）。以下是关于 value.deserializer 的详细信息：

value.deserializer=com.example.CustomValueDeserializer

value.deserializer 是 Kafka 生产者和消费者的一个关键配置选项，它定义了如何将 Kafka 消息的值反序列化为应用程序可以处理的格式。确保正确配置此选项对于生产者和消费者的正确运行非常重要。

bootstrap.servers 是 Kafka 生产者和消费者的一个重要配置选项，用于指定用于建立与 Kafka 集群的初始连接的主机/端口对列表。以下是关于 bootstrap.servers 的详细信息：

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

bootstrap.servers 是 Kafka 客户端配置中的一个关键选项，用于指定初始连接到 Kafka 集群的主机/端口对列表。正确配置此选项是建立与 Kafka 集群的初始连接的关键。

fetch.min.bytes 是 Kafka 消费者的一个配置选项，用于指定服务器在响应 fetch 请求之前应该返回的最小数据量。以下是关于 fetch.min.bytes 的详细信息：

fetch.min.bytes=100

fetch.min.bytes 是 Kafka 消费者配置中的一个选项，用于控制消费者在 fetch 请求中等待的最小数据量。配置此选项可以影响消费者的性能和延迟，可以根据具体的需求进行调整。

group.id 是 Kafka 消费者配置的一个重要选项，用于标识消费者属于哪个消费者组。以下是关于 group.id 的详细信息：

group.id=my-consumer-group

group.id 是 Kafka 消费者配置中的一个必要选项，用于标识消费者属于哪个消费者组。它对于实现消费者组的分布式消息处理非常重要。

heartbeat.interval.ms 是 Kafka 消费者配置的一个重要选项，它用于设置消费者向 Kafka 的消费者协调器发送心跳的时间间隔。以下是关于 heartbeat.interval.ms 的详细信息：

heartbeat.interval.ms=1000

heartbeat.interval.ms 用于控制消费者向 Kafka 消费者协调器发送心跳的频率，以维持消费者组的稳定性和协调工作。根据应用程序的需求，可以适当调整心跳间隔的值。

max.partition.fetch.bytes 是 Kafka 消费者配置的一个关键选项，用于控制每个分区从服务器获取的最大数据量。以下是关于 max.partition.fetch.bytes 的详细信息：

max.partition.fetch.bytes=524288

max.partition.fetch.bytes 是一个重要的 Kafka 消费者配置选项，用于控制每个分区从服务器获取的最大数据量。根据应用程序的需求和性能要求，可以调整这个值。

session.timeout.ms 是 Kafka 消费者配置的一个关键选项，用于检测客户端故障并触发 Kafka 消费者组中的重新平衡。以下是关于 session.timeout.ms 的详细信息：

session.timeout.ms=30000

session.timeout.ms 是 Kafka 消费者配置中的一个关键选项，用于检测客户端故障并触发重新平衡。根据应用程序的需求和性能要求，可以调整这个值，但必须确保它在代理服务器配置的允许范围内。

auto.offset.reset 是 Kafka 消费者配置的一个重要选项，用于确定当消费者在启动时没有初始偏移量，或者当前偏移量在服务器上不再存在时应该采取的操作。以下是关于 auto.offset.reset 的详细信息：

auto.offset.reset 允许配置消费者在没有初始偏移量或偏移量不存在时的行为。根据的需求，可以选择将消费者配置为从最早的消息、最新的消息开始消费，或者在没有偏移量信息时抛出异常。这取决于的数据消费策略和应用程序行为的要求。

enable.auto.commit 是 Kafka 消费者配置的一个选项，用于确定是否启用自动提交消费者的偏移量。以下是关于 enable.auto.commit 的详细信息：

enable.auto.commit 允许控制消费者是否应该在后台定期自动提交偏移量。可以根据应用程序的需求选择启用或禁用自动提交。如果启用，消费者将根据配置的时间间隔定期提交偏移量。如果禁用，需要手动管理偏移量的提交。

注:

Consumer Configs 查看更多配置

Kafka Connect Configs

配置如下：

config.storage.topic 是 Kafka 连接器配置的一个重要选项，用于指定存储连接器配置的 Kafka 主题的名称。以下是有关 config.storage.topic 的详细信息：

propertiesCopy code
config.storage.topic=my-connector-configs

config.storage.topic 是连接器配置的关键配置项，它定义了连接器配置存储在 Kafka 中的位置。确保将其设置为适当的 Kafka 主题，以便连接器可以正常工作并使用正确的配置。

group.id 是 Apache Kafka Connect 的一个重要配置项，用于唯一标识 Kafka Connect 集群中的工作节点（worker）。以下是关于 group.id 的详细信息：

propertiesCopy code
group.id=my-connect-cluster

group.id 是 Kafka Connect 集群中工作节点的唯一标识，确保每个工作节点都有不同的 group.id 以维持集群的正确运行。

key.converter 是 Kafka Connect 的一个关键配置项，用于指定用于将 Kafka Connect 格式与写入或读取 Kafka 的序列化形式之间进行转换的转换器类。这控制了写入或从 Kafka 读取的消息的键的格式，因为它独立于连接器，允许任何连接器与任何序列化格式一起工作。以下是有关 key.converter 的详细信息：

propertiesCopy code
key.converter=org.apache.kafka.connect.json.JsonConverter

key.converter 是 Kafka Connect 的一个关键配置项，它定义了如何将 Kafka Connect 数据格式转换为 Kafka 数据格式。确保选择与的数据格式和序列化格式兼容的转换器。

offset.storage.topic 是 Kafka Connect 的一个重要配置项，用于指定存储源连接器偏移量的 Kafka 主题的名称。这个配置项非常关键，因为它决定了连接器将偏移量信息写入或读取的 Kafka 主题。以下是有关 offset.storage.topic 的详细信息：

propertiesCopy code
offset.storage.topic=my-offsets

offset.storage.topic 是 Kafka Connect 的一个重要配置项，用于指定存储源连接器偏移量的 Kafka 主题的名称。确保该主题在 Kafka 中提前创建，并在配置中正确设置。这有助于确保连接器能够在重新启动时从上次停止的位置继续处理数据。

status.storage.topic 是 Kafka Connect 的一个关键配置项，用于指定存储连接器和任务状态信息的 Kafka 主题的名称。以下是有关 status.storage.topic 的详细信息：

propertiesCopy code
status.storage.topic=my-status

status.storage.topic 是 Kafka Connect 的一个关键配置项，用于指定存储连接器和任务状态信息的 Kafka 主题的名称。确保该主题在 Kafka 中提前创建，并在配置中正确设置。这有助于确保连接器和任务状态信息得以可靠地存储和监视。

value.converter 是 Kafka Connect 的一个关键配置项，用于指定连接器用于将 Kafka Connect 格式与写入或从 Kafka 读取的序列化形式之间进行转换的转换器类。这个配置项决定了写入或读取 Kafka 的消息中值的格式。以下是有关 value.converter 的详细信息：

propertiesCopy code
value.converter=org.apache.kafka.connect.json.JsonConverter

value.converter 是 Kafka Connect 的一个关键配置项，用于指定连接器用于将 Kafka Connect 格式的数据与写入或从 Kafka 读取的消息中的序列化数据之间进行转换的转换器类。确保选择与数据的序列化格式匹配的转换器类，并在配置中正确设置。这有助于确保数据正确地序列化和反序列化。

bootstrap.servers 是 Kafka 生产者和消费者的一个重要配置项，用于指定与 Kafka 集群建立初始连接的主机和端口对列表。以下是有关 bootstrap.servers 的详细信息：

propertiesCopy code
bootstrap.servers=broker1:9092,broker2:9092,broker3:9092

bootstrap.servers 是 Kafka 客户端配置中的关键项，用于指定初始连接到 Kafka 集群的服务器列表。配置正确的服务器列表是与 Kafka 集群建立连接的第一步，因此确保它正确设置非常重要。这有助于客户端发现并与整个 Kafka 集群进行通信。

注:

Kafka Connect Configs

Kafka Streams Configs

配置如下：

application.id 是用于流处理应用程序的标识符，它在 Kafka 集群内必须是唯一的。它在以下方面发挥作用：

application.id 是流处理应用程序的标识符，用于确保应用程序的 Kafka 客户端具有唯一的客户端 ID 和组 ID，并在状态存储中使用唯一的 changelog 主题前缀。这有助于在 Kafka 集群中正确管理和识别流处理应用程序。

num.standby.replicas 是指每个任务（task）的备用副本（standby replicas）数量。在 Kafka Streams 应用程序中，任务是处理数据流的单元，可以执行一些数据处理逻辑。备用副本是对任务的实时副本，它们可以用于故障恢复和提高容错性。

当一个任务正在执行并且其主要副本（primary replica）由于某种原因失败时，备用副本可以接管任务的处理。这有助于确保任务的处理不会因主要副本的故障而中断。

num.standby.replicas 参数用于配置备用副本的数量。通常，增加备用副本的数量可以提高容错性，因为它提供了更多的备用副本来接管任务，但这也会增加资源消耗。在配置时，你需要权衡容错性和资源消耗之间的权衡。

例如，如果将 num.standby.replicas 设置为 2，则每个任务将有一个主要副本和两个备用副本。这提供了更高的容错性，但也需要更多的资源来维护这些副本。

num.standby.replicas 参数允许你配置每个任务的备用副本数量，以提高流处理应用程序的容错性。

state.dir 参数指定了 Kafka Streams 应用程序的状态存储（state store）数据的目录位置。状态存储用于在流处理应用程序中维护和管理中间状态，以便进行数据处理和聚合。这些中间状态可以是键值对存储，帮助应用程序执行各种操作，如窗口操作、连接、过滤等。

每个 Kafka Streams 实例都需要一个唯一的状态存储目录，因为不同的实例可能在相同的文件系统上运行，并且需要避免命名冲突。

你可以通过设置 state.dir 参数来指定状态存储的目录位置。默认情况下，它的值是 /var/folders/6f/c4r3kvwd76l9c0k_59qzh_r00000gn/T//kafka-streams，但你可以根据你的需求将其更改为其他位置。确保每个 Kafka Streams 实例都有一个唯一的目录。

以下是一个示例配置：

propertiesCopy code
state.dir=/path/to/state/store/directory

在上面的配置中，你需要将 /path/to/state/store/directory 替换为你希望状态存储目录的实际路径。

state.dir 参数是用于指定 Kafka Streams 应用程序状态存储目录位置的重要配置参数。确保每个实例都有唯一的目录以避免冲突。

client.id 参数是一个用于 Kafka Streams 内部组件的客户端 ID 前缀字符串。它在内部消费者、生产者和还原消费者的客户端 ID 中使用。客户端 ID 的格式通常为 -StreamThread--。

这个参数通常用于标识与 Kafka Streams 应用程序相关的各个内部组件，以便在 Kafka 服务器日志中进行诊断和监视。设置不同的 client.id 前缀可以帮助你区分不同的组件。

虽然 client.id 参数的默认值为空字符串 ""，但你可以根据需要将其设置为一个有意义的值，以便更好地识别和监视你的 Kafka Streams 应用程序的内部组件。

以下是一个示例配置：

propertiesCopy code
client.id=my-streams-app

在上面的配置中，client.id 被设置为 my-streams-app，这将成为内部组件客户端 ID 的前缀。

client.id 参数用于设置 Kafka Streams 应用程序内部组件的客户端 ID 前缀，以帮助你识别和监视这些组件。你可以根据需要为其指定一个有意义的值。

注:

Kafka Streams Configs

Admin Configs

配置如下：

connections.max.idle.ms 参数用于配置在连接处于空闲状态多少毫秒后关闭这些连接。连接的空闲状态是指在没有任何网络活动的情况下连接保持打开状态的时间。

默认情况下，connections.max.idle.ms 的值为 300,000 毫秒，即 5 分钟。这意味着如果连接在 5 分钟内没有任何活动，它将被自动关闭。

你可以根据你的应用程序的需求来调整这个值。如果你的应用程序有一些长时间不使用的连接，你可以增加这个值以减少连接的频繁打开和关闭。另一方面，如果你希望尽快释放不再需要的连接以释放资源，你可以减少这个值。

根据你的网络环境和应用程序的工作负载，调整 connections.max.idle.ms 可能会对性能产生影响，因此建议进行测试以找到最适合你的场景的值。

default.api.timeout.ms 参数用于指定客户端 API 的超时时间，以毫秒为单位。这个配置作为所有未指定超时参数的客户端操作的默认超时时间。

• 参数名称：default.api.timeout.ms
• 默认值：60000 毫秒（1 分钟）
• 有效值范围：大于或等于 0 的整数
• 重要性：中等

这个参数是全局性的，它会影响所有的客户端操作，包括生产者、消费者、管理操作等。当你执行一个客户端操作而没有显式地指定超时时间时，将会使用 default.api.timeout.ms 作为该操作的超时时间。

超时时间是指客户端等待服务器响应的最大时间。如果在超时时间内未收到服务器响应，客户端将抛出超时异常。超时时间的设置对于确保客户端操作不会无限期地等待响应非常重要，同时也会影响客户端的响应速度。

• default.api.timeout.ms 参数用于配置客户端 API 的默认超时时间。
• 默认情况下，超时时间为 1 分钟，可以根据需要进行调整。
• 超时时间是客户端等待服务器响应的最大时间，对于确保客户端操作不会无限期地等待响应非常重要。
• 这个配置是全局性的，会影响所有的客户端操作，包括生产者、消费者、管理操作等。

request.timeout.ms 参数用于控制客户端等待请求响应的最大时间。如果在超时时间内未收到响应，客户端将会采取重新发送请求或者在重试耗尽后失败请求的方式处理。

• 参数名称：request.timeout.ms
• 默认值：30000 毫秒（30 秒）
• 有效值范围：大于或等于 0 的整数
• 重要性：中等

这个参数对于客户端与 Kafka 服务器之间的通信非常重要。它决定了客户端在发送请求后等待响应的最大时间。如果响应没有在超时时间内到达，客户端将会采取以下两种行为之一：

request.timeout.ms 的值可以根据应用程序的需求进行调整。较短的超时时间可能会导致更快的失败响应，但可能增加了请求失败的风险，而较长的超时时间可以更容忍网络不稳定或响应时间较长的情况，但可能会导致客户端等待的时间增加。

注:

Admin Configs

System Properties

配置如下：

org.apache.kafka.disallowed.login.modules 是一个系统属性，用于禁用 SASL JAAS 配置中的问题登录模块（login modules）的使用。这个属性接受逗号分隔的登录模块名称列表。默认情况下，com.sun.security.auth.module.JndiLoginModule 登录模块被禁用。

如果用户希望启用 JndiLoginModule，用户需要显式地重置系统属性，如下所示。我们建议用户验证配置，并只允许受信任的 JNDI 配置。有关更多详细信息，请参考 CVE-2023-25194。

diffCopy code
-Dorg.apache.kafka.disallowed.login.modules=

要禁用更多的登录模块，请更新系统属性，使用逗号分隔的登录模块名称列表。请确保将 JndiLoginModule 模块名称显式添加到逗号分隔的列表中，如下所示：

arduinoCopy code
-Dorg.apache.kafka.disallowed.login.modules=com.sun.security.auth.module.JndiLoginModule,com.ibm.security.auth.module.LdapLoginModule,com.ibm.security.auth.module.Krb5LoginModule

这个属性自 Kafka 版本 3.4.0 开始提供，默认情况下禁用了 com.sun.security.auth.module.JndiLoginModule 登录模块。这个属性的存在允许管理员限制哪些登录模块可以在 Kafka 中使用，以提高安全性。

注：kafka得配置太多，其实感觉能用到得都不是很多，需要的话，就去官网查看对应得配置，并查阅相关资料即可。

总结

以上内容来自Kafka官方文档的学习，我们可以从中了解到kafka相关的知识，以及它相关的配置信息。后面，我们来了解kafka的设计和kafka的API具体使用。