【Docker】Kafka 容器部署

Kafka docker 容器化部署

Kafka标准软件基于Bitnami Kafka 构建。当前版本为3.6.1

你可以通过轻云UC部署工具直接安装部署，也可以手动按如下文档操作，该项目已经全面开源，可以从如下环境获取
配置文件地址: https://gitee.com/qingplus/qingcloud-platform

简介

Kafka是一种高吞吐量的分布式发布订阅消息系统（消息引擎系统），它是最初由Linkedin公司开发，是一个分布式、支持分区的（partition）、多副本的（replica），它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟的实时系统、Storm/Spark流式处理引擎，web/nginx日志、访问日志，消息服务等等，用scala语言编写，Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目。

配置

可以使用以下环境变量通过 Apache Kafka Docker 设置配置：

KAFKA_CERTIFICATE_PASSWORD：证书的密码。没有默认值。
KAFKA_HEAP_OPTS：Apache Kafka 的 Java 堆大小。默认值：-Xmx1024m -Xms1024m。
KAFKA_ZOOKEEPER_PROTOCOL：Zookeeper 连接的身份验证协议。允许的协议：PLAINTEXT、SASL、SSL和SASL_SSL。默认值：纯文本。
KAFKA_ZOOKEEPER_USER：用于 SASL 身份验证的 Apache Kafka Zookeeper 用户。没有默认值。
KAFKA_ZOOKEEPER_PASSWORD：用于 SASL 身份验证的 Apache Kafka Zookeeper 用户密码。没有默认值。
KAFKA_ZOOKEEPER_TLS_KEYSTORE_PASSWORD：Apache Kafka Zookeeper密钥库文件密码和密钥密码。没有默认值。
KAFKA_ZOOKEEPER_TLS_TRUSTSTORE_PASSWORD：Apache Kafka Zookeeper 信任库文件密码。没有默认值。
KAFKA_ZOOKEEPER_TLS_VERIFY_HOSTNAME：验证 TLS 证书上的 Zookeeper 主机名。默认值：true。
KAFKA_ZOOKEEPER_TLS_TYPE：选择要使用的 TLS 证书格式。允许值：JKS, PEM. 默认值：JKS。
KAFKA_CFG_LISTENERS：Kafkalisteners配置覆盖。没有默认值。
KAFKA_CFG_ADVERTISED_LISTENERS：Kafkaadvertised.listeners配置覆盖。没有默认值。
KAFKA_CFG_SASL_ENABLED_MECHANISMS：将 SASL 用于客户端、代理间或 Zookeeper 通信时允许的机制。允许的值：PLAIN、SCRAM-SHA-256或SCRAM-SHA-512这些值的逗号分隔组合。默认值：PLAIN、SCRAM-SHA-256、SCRAM-SHA-512。注意：KRaft <= 3.4 尚不支持 SCRAM 机制，因此此列表将自动缩减为仅PLAINSASL 机制。
KAFKA_TLS_CLIENT_AUTH：设置的值ssl.client.auth。允许值：required, requested, none. 默认值：必填。
KAFKA_TLS_<uppercase_listener_name>_CLIENT_AUTHlistener.name.<listener>.ssl.client.auth：设置用于通过 SASL 配置 mTLS的值。允许值：required, requested, none.
KAFKA_TLS_TYPE：选择要使用的 TLS 证书格式。允许值：JKS, PEM. 默认值：JKS。
KAFKA_CLIENT_USERS：使用 SASL 进行客户端通信时将在 Zookeeper 中创建的用户。被逗号隔开。默认：用户
KAFKA_CLIENT_PASSWORDS：在处指定的用户的密码KAFKA_CLIENT_USERS。被逗号隔开。默认值：bitnami
KAFKA_CFG_MAX_PARTITION_FETCH_BYTES：服务器将返回的每个分区的最大数据量。没有默认值。
KAFKA_CFG_MAX_REQUEST_SIZE：请求的最大大小（以字节为单位）。没有默认值。
KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL：用于代理间通信的 SASL 机制。没有默认值。注意：KRaft <= 3.4 尚不支持 SCRAM 机制，因此 KRaft 模式下唯一支持的 SASL 机制是PLAIN.
KAFKA_INTER_BROKER_USER：Apache Kafka 代理间通信用户。默认值：用户。
KAFKA_INTER_BROKER_PASSWORD：Apache Kafka 代理间通信密码。默认值：bitnami。
KAFKA_CFG_SASL_MECHANISM_CONTROLLER_PROTOCOL：用于控制器通信的 SASL 机制。没有默认值。注意：KRaft <= 3.4 尚不支持 SCRAM 机制，因此 KRaft 模式下唯一支持的 SASL 机制是PLAIN.
KAFKA_CONTROLLER_USER：Apache Kafka 控制器通信用户。默认值：controller_user。
KAFKA_CONTROLLER_PASSWORD：Apache Kafka控制器通信密码。默认值：bitnami。
KAFKA_CFG_PROCESS_ROLES：KRaft 模式下运行时的节点角色。没有默认值。
KAFKA_CFG_NODE_ID：唯一的节点id，在KRaft模式下运行时需要。没有默认值。
KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP：使用 Apache Kafka 安全协议映射每个侦听器。如果节点设置了controller角色，则需要此设置才能为CONTROLLER LISTENER. 例如：PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT。没有默认值。
KAFKA_CFG_CONTROLLER_QUORUM_VOTERS：以逗号分隔的 {id}@{host}:{port} 条目列表中的控制器仲裁投票者集的 id/端点信息映射。没有默认值。
KAFKA_CFG_CONTROLLER_LISTENER_NAMES：控制器使用的侦听器名称的逗号分隔列表。如果在 KRaft 模式下运行，这是必需的。没有默认值。
KAFKA_KRAFT_CLUSTER_ID：使用Kafka Raft（KRaft）时的Kafka集群ID。没有默认值。
此外，任何以KAFKA_CFG_开头的环境变量都将映射到其相应的 Apache Kafka 键。例如，使用KAFKA_CFG_BACKGROUND_THREADS设置background.threads或KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE配置auto.create.topics.enable。

docker run --name kafka 
        -e KAFKA_CFG_PROCESS_ROLES ... 
        -e KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=true 
        registry.cn-hangzhou.aliyuncs.com/qingcloudtech/kafka:latest

或者通过修改docker-compose.yml:

kafka:
  ...
  environment:
    - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=true
  ...

Apache Kafka 开发设置示例

要在开发设置中使用 Apache Kafka，请创建以下docker-compose.yml文件：

version: "3"
services:
  kafka:
    image: 'registry.cn-hangzhou.aliyuncs.com/qingcloudtech/kafka:latest'
    ports:
      - '9092:9092'
    environment:
      - KAFKA_CFG_NODE_ID=0
      - KAFKA_CFG_PROCESS_ROLES=controller,broker
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER

要部署它，请在文件所在目录中运行以下命令docker-compose.yml：

docker-compose up -d

kafka 与zookeeper配置

Apache Kafka Raft (KRaft) 在 Kafka 中使用了新的仲裁控制器服务，该服务取代了之前的控制器，并使用了 Raft 共识协议的基于事件的变体。这极大地简化了 Kafka 的架构，将元数据整合到 Kafka 本身，而不是将其拆分到两个不同的系统：ZooKeeper 和 Kafka。

如果您想继续使用ZooKeeper，可以使用以下配置：

version: "3.9"

services:
  zookeeper:
    image: registry.cn-hangzhou.aliyuncs.com/qingcloudtech/zookeeper:3.9
    ports:
      - "2181:2181"
    volumes:
      - "zookeeper_data:/bitnami"
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes
  kafka:
    image: registry.cn-hangzhou.aliyuncs.com/qingcloudtech/kafka:3.6.1
    ports:
      - "9092:9092"
    volumes:
      - "kafka_data:/bitnami"
    environment:
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
    depends_on:
      - zookeeper
volumes:
  zookeeper_data:
    driver: local
  kafka_data:
    driver: local

使用内部和外部客户端访问 Apache Kafka

为了使用内部和外部客户端访问 Apache Kafka 代理，您需要为每个客户端配置一个侦听器。

为此，请将以下环境变量添加到 docker-compose 中：

    environment:
      - KAFKA_CFG_NODE_ID=0
      - KAFKA_CFG_PROCESS_ROLES=controller,broker
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@<your_host>:9093
+     - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094
+     - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092,EXTERNAL://localhost:9094
+     - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER

并暴露外部端口：

内部客户端仍然可以在docker网络内使用

    ports:
-     - '9092:9092'
+     - '9094:9094'

注意：要从外部进行连接，请将localhost上面的内容更改为主机的外网IP/主机名并包含EXTERNAL://0.0.0.0:9093在内KAFKA_CFG_LISTENERS以允许远程连接。

使用外部客户端的生产者和消费者

这些来自同一主机的客户端将用于localhost连接到 Apache Kafka。

kafka-console-producer.sh --producer.config /opt/bitnami/kafka/config/producer.properties --bootstrap-server 127.0.0.1:9094 --topic test
kafka-console-consumer.sh --consumer.config /opt/bitnami/kafka/config/consumer.properties --bootstrap-server 127.0.0.1:9094 --topic test --from-beginning

如果从另一台计算机运行这些命令，请相应地更改地址。

使用内部客户端的生产者和消费者

这些来自同一 Docker 网络上其他容器的客户端将使用 kafka 容器服务主机名连接到 Apache Kafka。

kafka-console-producer.sh --producer.config /opt/bitnami/kafka/config/producer.properties --bootstrap-server kafka:9092 --topic test
kafka-console-consumer.sh --consumer.config /opt/bitnami/kafka/config/consumer.properties --bootstrap-server kafka:9092 --topic test --from-beginning

同样，应用程序代码需要使用bootstrap.servers=kafka:9092

安全

为了配置身份验证，您必须正确配置 Apache Kafka 侦听器。让我们看一个示例，为 Apache Kafka 配置SASL_SSL与客户端通信的身份验证以及SASL与控制器相关的通信的身份验证。

定义以下环境变量，以及用于客户端通信的 SASL 凭据：

KAFKA_CFG_LISTENERS=SASL_SSL://:9092,CONTROLLER://:9093
KAFKA_CFG_ADVERTISED_LISTENERS=SASL_SSL://localhost:9092
KAFKA_CLIENT_USERS=user
KAFKA_CLIENT_PASSWORDS=password
KAFKA_CLIENT_LISTENER_NAME=SASL_SSL
KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
KAFKA_CFG_SASL_MECHANISM_CONTROLLER_PROTOCOL=PLAIN
KAFKA_CONTROLLER_USER=controller_user
KAFKA_CONTROLLER_PASSWORD=controller_password

必须使用自己的 SSL 证书。您可以将 Java 密钥存储或 PEM 文件放入/opt/bitnami/kafka/config/certs. 如果 JKS 或 PEM 证书受密码保护（推荐），您将需要提供它才能访问密钥库：

KAFKA_CERTIFICATE_PASSWORD=myCertificatePassword

如果相关证书安装在/opt/bitnami/kafka/config/certs/kafka.truststore.jks、/opt/bitnami/kafka/config/certs/kafka.truststore.pem，/bitnami/kafka/config/certs/kafka.truststore.jks，/bitnami/kafka/config/certs/kafka.truststore.pem，请设置该KAFKA_TLS_TRUSTSTORE_FILE变量。

以下脚本可以帮助您创建 JKS 和证书：

kafka-generate-ssl.sh

请记住以下注意事项：
当提示输入密码时，请为所有节点使用相同的密码。

将 Common Name 或 FQDN 值设置Apache Kafka 容器主机名，例如kafka.example.com。输入此值后，当提示“您的名字和姓氏是什么？”时，也输入此值。
作为替代方案，您可以禁用主机名验证，将环境变量设置KAFKA_CFG_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM为空字符串。
设置 Apache Kafka 集群时（请查看“设置 Apache Kafka 集群”以获取更多信息），每个 Apache Kafka 代理和逻辑客户端都需要自己的密钥库。您必须对集群中的每个代理重复该过程。
以下 docker-compose 文件是一个示例，显示如何挂载受密码保护的 JKS 证书certificatePassword123。此外，它还指定 Apache Kafka 容器主机名以及客户端和 Zookeeper 用户的凭据。

version: '2'

services:
  kafka:
    image: 'registry.cn-hangzhou.aliyuncs.com/qingcloudtech/kafka:latest'
    hostname: kafka.example.com
    ports:
      - '9092'
    environment:
      - KAFKA_CFG_NODE_ID=0
      - KAFKA_CFG_PROCESS_ROLES=controller,broker
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093
      - KAFKA_CFG_LISTENERS=SASL_SSL://:9092,CONTROLLER://:9093
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
      - KAFKA_CFG_ADVERTISED_LISTENERS=SASL_SSL://:9092
      - KAFKA_CLIENT_USERS=user
      - KAFKA_CLIENT_PASSWORDS=password
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_SASL_MECHANISM_CONTROLLER_PROTOCOL=PLAIN
      - KAFKA_CONTROLLER_USER=controller_user
      - KAFKA_CONTROLLER_PASSWORD=controller_password
      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=SASL_SSL
      - KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL=PLAIN
      - KAFKA_INTER_BROKER_USER=controller_user
      - KAFKA_INTER_BROKER_PASSWORD=controller_password
      - KAFKA_CERTIFICATE_PASSWORD=certificatePassword123
      - KAFKA_TLS_TYPE=JKS # or PEM
    volumes:
      # Both .jks and .pem files are supported
      # - './kafka.keystore.pem:/opt/bitnami/kafka/config/certs/kafka.keystore.pem:ro'
      # - './kafka.keystore.key:/opt/bitnami/kafka/config/certs/kafka.keystore.key:ro'
      # - './kafka.truststore.pem:/opt/bitnami/kafka/config/certs/kafka.truststore.pem:ro'
      - './kafka.keystore.jks:/opt/bitnami/kafka/config/certs/kafka.keystore.jks:ro'
      - './kafka.truststore.jks:/opt/bitnami/kafka/config/certs/kafka.truststore.jks:ro'

为了获得使用和生成消息所需的凭据，您需要在客户端中提供凭据。如果您的 Apache Kafka 客户端允许，请使用您提供的凭据。

在使用bitnami/kafka image生成和使用消息时，您需要指向consumer.properties和/或producer.properties文件，其中包含工作所需的配置。您可以在该目录中找到该文件/opt/bitnami/kafka/config。

生成消息：

kafka-console-producer.sh --bootstrap-server 127.0.0.1:9092 --topic test --producer.config /opt/bitnami/kafka/config/producer.properties

消费消息：

kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic test --consumer.config /opt/bitnami/kafka/config/consumer.properties

使用其他工具来使用 Apache Kafka 集群，则需要提供所需的信息。您可以在位于目录的文件中找到所需的信息/opt/bitnami/kafka/config。

Inter-Broker 通信

当部署具有多个代理的 Apache Kafka 集群时，可以使用以下变量SASL配置代理间通信：SASL_SSL

KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL：Apache Kafka 代理间通信协议。
KAFKA_INTER_BROKER_USER：Apache Kafka 代理间通信用户。
KAFKA_INTER_BROKER_PASSWORD：Apache Kafka 代理间通信密码。
注意：在 KRaft 模式下运行时，KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL 仅支持PLAINKafka 版本 <= 3.4 中的机制。

Control plane通信

在 KRaft 模式下部署具有多个控制器的 Apache Kafka 集群时，可以使用以下变量SASL配置控制器通信：SASL_SSL

KAFKA_CFG_SASL_MECHANISM_CONTROLLER_PROTOCOL：Apache Kafka 控制器通信协议。
KAFKA_CONTROLLER_USER：Apache Kafka 控制器通信用户。目前仅PLAIN支持机制。
KAFKA_CONTROLLER_PASSWORD：Apache Kafka控制器通信密码。
注意：在KRaft模式下运行时，KAFKA_CFG_SASL_MECHANISM_CONTROLLER_PROTOCOL仅支持PLAIN机制。

Apache Kafka SASL 配置

配置 Apache Kafka 侦听器SASL或SASL_SSL与客户端通信时，您可以使用以下环境变量提供 SASL 凭据：

KAFKA_CLIENT_USERS：Apache Kafka 客户端用户。默认：user
KAFKA_CLIENT_PASSWORDS：Apache Kafka 客户端用户密码。默认值：bitnami
注意：在 KRaft 模式下运行时，只有第一个user:password对才会生效，因为 KRaft 模式尚不支持 SCRAM 机制。

Apache Kafka KRaft 模式配置

可以通过提供以下值来启用 KRaft 模式：

KAFKA_CFG_PROCESS_ROLES：以逗号分隔的 Kafka KRaft 角色列表。允许值：‘controller,broker’, ‘controller’, ’broker‘.
KAFKA_CFG_NODE_ID：Kafka节点的唯一id。
KAFKA_CFG_LISTENERS：Kafka 监听器列表。如果节点设置了角色包含controller，则必须包含CONTROLLER监听器。
KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP：使用 Apache Kafka 安全协议映射每个侦听器。如果节点设置了controller角色，则需要此设置环境变量. 例如：PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT。
为了在没有身份验证的情况下配置控制器通信，应该提供以下环境变量：
KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: 应包括CONTROLLER:PLAINTEXT.

为了配置 Apache Kafka 控制器通信带SASL，应该提供以下环境变量：

KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: 应包括CONTROLLER:SASL.
KAFKA_CFG_SASL_MECHANISM_CONTROLLER_PROTOCOL：用于控制器通信的 SASL 机制。注意：KRaft 模式尚不支持 SCRAM 机制，因此 KRaft 模式中唯一支持的 SASL 机制是PLAIN.
KAFKA_CONTROLLER_USER：Apache Kafka 控制器通信用户。
KAFKA_CONTROLLER_PASSWORD：Apache Kafka控制器通信密码。

为了配置 Apache Kafka 控制器通信带SSL，应该提供以下环境变量：

KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: 应包括CONTROLLER:SSL.
KAFKA_TLS_<uppercase_controller_listener_name>_CLIENT_AUTH：配置 kafka 控制平面通信的 mTLS 身份验证方法。允许值：required, requested, none.
KAFKA_TLS_TYPE：选择要使用的 TLS 证书格式。允许值：JKS, PEM. 默认值：JKS。
有效的密钥库和信任库安装在/opt/bitnami/kafka/config/certs/kafka.keystore.jks和处/opt/bitnami/kafka/config/certs/kafka.truststore.jks。
为了使用 Zookeeper 服务器对 Apache Kafka 进行身份验证SASL_SSL，应该提供以下环境变量：
KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: 应包括CONTROLLER:SASL_SSL.
KAFKA_CFG_SASL_MECHANISM_CONTROLLER_PROTOCOL：用于控制器通信的 SASL 机制。注意：KRaft 模式尚不支持 SCRAM 机制，因此 KRaft 模式中唯一支持的 SASL 机制是PLAIN.
KAFKA_CONTROLLER_USER：Apache Kafka 控制器通信用户。
KAFKA_CONTROLLER_PASSWORD：Apache Kafka控制器通信密码。
KAFKA_TLS_<uppercase_controller_listener_name>_CLIENT_AUTH：配置 kafka 控制平面通信的 mTLS 身份验证方法。允许值：required, requested, none.
KAFKA_TLS_TYPE：选择要使用的 TLS 证书格式。允许值：JKS, PEM. 默认值：JKS。
有效的密钥库和信任库安装在/opt/bitnami/kafka/config/certs/kafka.keystore.jks和处/opt/bitnami/kafka/config/certs/kafka.truststore.jks。
SSL注意：SSL 设置由使用或协议配置的所有侦听器共享SASL_SSL。尚不支持为每个侦听器设置不同的证书。

Apache Kafka ZooKeeper 模式配置

可以通过提供以下值来启用 Zookeeper 模式：

KAFKA_CFG_ZOOKEEPER_CONNECT：以逗号分隔的 Zookeeper 连接字符串列表。例如<zk_host1>:<zk_port1>,<zk_host2>:<zk_port2>
KAFKA_CFG_BROKER_ID：Kafka Broker 的可选ID。如果不设置，将自动生成一个随机ID。
连接 Zookeeper 服务器有不同的配置选项。

为了无需身份验证即可连接 Zookeeper 服务器，您应该提供以下环境变量：

KAFKA_ZOOKEEPER_PROTOCOL：纯文本。

为了使用 Zookeeper 服务器对 Apache Kafka 进行身份验证SASL，您应该提供以下环境变量：

KAFKA_ZOOKEEPER_PROTOCOL：SASL。
KAFKA_ZOOKEEPER_USER：用于 SASL 身份验证的 Apache Kafka Zookeeper 用户。没有默认值。
KAFKA_ZOOKEEPER_PASSWORD：用于 SASL 身份验证的 Apache Kafka Zookeeper 用户密码。没有默认值。

为了使用 Zookeeper 服务器对 Apache Kafka 进行身份验证SSL，您应该提供以下环境变量：

KAFKA_ZOOKEEPER_PROTOCOL：SSL。
KAFKA_ZOOKEEPER_TLS_KEYSTORE_PASSWORD：Apache Kafka Zookeeper密钥库文件密码和密钥密码。没有默认值。
KAFKA_ZOOKEEPER_TLS_TRUSTSTORE_PASSWORD：Apache Kafka Zookeeper 信任库文件密码。没有默认值。
KAFKA_ZOOKEEPER_TLS_VERIFY_HOSTNAME：验证 TLS 证书上的 Zookeeper 主机名。默认值：true。
KAFKA_ZOOKEEPER_TLS_TYPE：选择要使用的 TLS 证书格式。允许值：JKS, PEM. 默认值：JKS。

为了使用 Zookeeper 服务器对 Apache Kafka 进行身份验证SASL_SSL，您应该提供以下环境变量：

KAFKA_ZOOKEEPER_PROTOCOL：SASL_SSL。
KAFKA_ZOOKEEPER_USER：用于 SASL 身份验证的 Apache Kafka Zookeeper 用户。没有默认值。
KAFKA_ZOOKEEPER_PASSWORD：用于 SASL 身份验证的 Apache Kafka Zookeeper 用户密码。没有默认值。
KAFKA_ZOOKEEPER_TLS_TRUSTSTORE_FILE：Apache Kafka Zookeeper 信任库文件位置。如果安装位置不同于/bitnami/kafka/conf/certs/zookeeper.truststore.pem、/bitnami/kafka/conf/certs/zookeeper.truststore.jks或无默认值，请设置/opt/bitnami/kafka/config/certs/zookeeper.truststore.jks它。/opt/bitnami/kafka/conf/certs/zookeeper.truststore.pem
KAFKA_ZOOKEEPER_TLS_KEYSTORE_PASSWORD：Apache Kafka Zookeeper密钥库文件密码和密钥密码。没有默认值。
KAFKA_ZOOKEEPER_TLS_TRUSTSTORE_PASSWORD：Apache Kafka Zookeeper 信任库文件密码。没有默认值。
KAFKA_ZOOKEEPER_TLS_VERIFY_HOSTNAME：验证 TLS 证书上的 Zookeeper 主机名。默认值：true。
KAFKA_ZOOKEEPER_TLS_TYPE：选择要使用的 TLS 证书格式。允许值：JKS, PEM. 默认值：JKS。
注意：您还必须使用自己的 SSL 证书。您可以将 Java 密钥库（zookeeper.keystore.jks和）或zookeeper.truststore.jksPEM 文件（zookeeper.keystore.pem和）安装到zookeeper.keystore.key. 如果客户端身份验证是在Zookeeper 中，则证书文件是可选的。zookeeper.truststore.pem/opt/bitnami/kafka/conf/certsnonewant

设置 Apache Kafka 集群

可以使用以下环境变量通过设置 Apache Kafka 集群：

KAFKA_CFG_CONTROLLER_QUORUM_VOTERS：逗号分隔的host:port对，每个对应一个 Kafka 控制器连接。

步骤 1：为 Apache Kafka 创建第一个节点

第一步是创建一个 Apache Kafka 实例。

docker run --name kafka-0 \
  --network app-tier \
  -e KAFKA_CFG_NODE_ID=0 \
  -e KAFKA_CFG_PROCESS_ROLES=controller,broker \
  -e KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
  -e KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
  -e KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka-0:9093,1@kafka-1:9093,2@kafka-2:9093 \
  -e KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR=3 \
  -e KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=3 \
  -e KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR=2 \
  -e KAFKA_KRAFT_CLUSTER_ID=abcdefghijklmnopqrstuv \
  -p :9092 \
  -p :9093 \
  registry.cn-hangzhou.aliyuncs.com/qingcloudtech/kafka:latest

步骤 2：创建第二个节点

接下来我们启动一个新的 Apache Kafka 容器。

docker run --name kafka-1 \
  --network app-tier \
  -e KAFKA_CFG_NODE_ID=1 \
  -e KAFKA_CFG_PROCESS_ROLES=controller,broker \
  -e KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
  -e KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
  -e KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka-0:9093,1@kafka-1:9093,2@kafka-2:9093 \
  -e KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR=3 \
  -e KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=3 \
  -e KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR=2 \
  -e KAFKA_KRAFT_CLUSTER_ID=abcdefghijklmnopqrstuv \
  -p :9092 \
  -p :9093 \
  registry.cn-hangzhou.aliyuncs.com/qingcloudtech/kafka:latest

步骤3：创建第三个节点

接下来我们启动另一个新的 Apache Kafka 容器。

docker run --name kafka-3 \
  --network app-tier \
  -e KAFKA_CFG_NODE_ID=3 \
  -e KAFKA_CFG_PROCESS_ROLES=controller,broker \
  -e KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
  -e KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
  -e KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka-0:9093,1@kafka-1:9093,2@kafka-2:9093 \
  -e KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR=3 \
  -e KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=3 \
  -e KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR=2 \
  -e KAFKA_KRAFT_CLUSTER_ID=abcdefghijklmnopqrstuv \
  -p :9092 \
  -p :9093 \
  registry.cn-hangzhou.aliyuncs.com/qingcloudtech/kafka:latest

一个 Apache Kafka 集群配置完成。可以通过添加/删除从属服务器来扩展集群，而不会导致任何停机。

可以在文件中找到此部署的 docker-compose 版本docker-compose-cluster.yml。

示例：创建一个 replicated topic

可以使用以下命令创建：

root@kafka-0:/# /opt/bitnami/kafka/bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --topic mytopic --partitions 3 --replication-factor 3
Created topic "mytopic".

root@kafka-0:/# /opt/bitnami/kafka/bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic mytopic
Topic:mytopic   PartitionCount:3        ReplicationFactor:3     Configs:
Topic: mytopic  Partition: 0    Leader: 2       Replicas: 2,3,1 Isr: 2,3,1
Topic: mytopic  Partition: 1    Leader: 3       Replicas: 3,1,2 Isr: 3,1,2
Topic: mytopic  Partition: 2    Leader: 1       Replicas: 1,2,3 Isr: 1,2,3

All配置

该镜像中所有配置可以在/bitnami/kafka/config/中查找，可以通过设置 KAFKA_MOUNTED_CONF_DIR 环境变量来更改该位置。

docker run --name kafka -v /path/to/server.properties:/bitnami/kafka/config/server.properties registry.cn-hangzhou.aliyuncs.com/qingcloudtech/kafka:latest

之后，更改将被同步到容器中。

第 1 步：运行 Apache Kafka 映像

运行 Apache Kafka 映像，从主机安装目录。

修改docker-compose.yml文件：

...
services:
  kafka:
    ...
    volumes:
      - 'kafka_data:/bitnami'
+     - /path/to/server.properties:/bitnami/kafka/config/server.properties

第 2 步：编辑配置

使用您喜欢的编辑器编辑主机上的配置。

vi /path/to/server.properties

第3步：重启Apache Kafka
更改配置后，重新启动 Apache Kafka 容器以使更改生效。

docker restart kafka

或者使用 Docker Compose：

docker-compose restart kafka

日志
容器日志发送到stdout. 要查看日志：

docker logs kafka

或者使用 Docker Compose：

docker-compose logs kafka

维护

备份您的容器

要备份数据、配置和日志，请按照以下简单步骤操作：

第 1 步：停止当前正在运行的容器

docker stop kafka

或者使用 Docker Compose：

docker-compose stop kafka

步骤2：运行备份命令

我们需要在用于创建备份的容器中安装两个卷：主机上用于存储备份的目录，以及我们刚刚停止的容器中的卷，以便我们可以访问数据。

docker run --rm -v /path/to/kafka-backups:/backups --volumes-from kafka busybox \
cp -a /bitnami/kafka /backups/latest

或者使用 Docker Compose：

docker run --rm -v /path/to/kafka-backups:/backups --volumes-from `docker-compose ps -q kafka` busybox \
cp -a /bitnami/kafka /backups/latest

恢复备份

恢复备份就像将备份作为卷安装在容器中一样简单。

docker run -v /path/to/kafka-backups/latest:/bitnami/kafka registry.cn-hangzhou.aliyuncs.com/qingcloudtech/kafka:latest

修改docker-compose.yml文件：

kafka:
  volumes:
    - /path/to/kafka-backups/latest:/bitnami/kafka

从 Zookeeper 模式迁移到 KRaft 模式

本节介绍 Kafka 从 Zookeeper 模式到 KRaft 模式的迁移。

1.从 Zookeeper 获取cluster ID

2.配置controller节点,添加以下环境变量

KAFKA_CFG_PROCESS_ROLES=controller
KAFKA_CFG_NODE_ID=<unique_id>
KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=<controller1_node_id>@<controller1_host>:9093,<controller2_node_id>@<controller2_host>:9093,...
KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
KAFKA_CFG_LISTENERS=CONTROLLER://:9093
KAFKA_CFG_ZOOKEEPER_METADATA_MIGRATION_ENABLE=true
KAFKA_CFG_ZOOKEEPER_CONNECT=<zk_host>:<zk_port>
KAFKA_KRAFT_CLUSTER_ID=<cluster_id_step1>

3.使用迁移设置配置代理：

KAFKA_CFG_BROKER_ID=<current_broker_id>
KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=<controller1_node_id>@<controller1_host>:9093,<controller2_node_id>@<controller2_host>:9093,...
KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
KAFKA_CFG_INTER_BROKER_PROTOCOL_VERSION=3.4
KAFKA_CFG_ZOOKEEPER_METADATA_MIGRATION_ENABLE=true
KAFKA_CFG_ZOOKEEPER_CONNECT=<zk_host>:<zk_port>

4.迁移代理：

KAFKA_CFG_PROCESS_ROLES=broker
KAFKA_CFG_NODE_ID=<unique_id>
KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=<controller1_node_id>@<controller1_host>:9093,<controller2_node_id>@<controller2_host>:9093,...
KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER

5.禁用控制器上的迁移模式：

KAFKA_CFG_PROCESS_ROLES=controller
KAFKA_CFG_NODE_ID=<unique_id>
KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=<controller1_node_id>@<controller1_host>:9093,<controller2_node_id>@<controller2_host>:9093,...
KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
KAFKA_CFG_LISTENERS=CONTROLLER://:9093
KAFKA_KRAFT_CLUSTER_ID=<cluster_id_step1>

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 223,689评论 6赞 521
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 95,685评论 3赞 401
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 170,676评论 0赞 366
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 60,496评论 1赞 300
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 69,512评论 6赞 399
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 53,044评论 1赞 314
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 41,423评论 3赞 427
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 40,399评论 0赞 278
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 46,923评论 1赞 323
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 38,973评论 3赞 343
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 41,117评论 1赞 354
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 36,761评论 5赞 350
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 42,440评论 3赞 336
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 32,929评论 0赞 25
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 34,045评论 1赞 275
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 49,609评论 3赞 380
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 46,147评论 2赞 363

【Docker】Kafka 容器部署

Kafka docker 容器化部署

简介

配置

Apache Kafka 开发设置示例

kafka 与zookeeper配置

使用内部和外部客户端访问 Apache Kafka

使用外部客户端的生产者和消费者

使用内部客户端的生产者和消费者

安全

Inter-Broker 通信

Control plane通信

Apache Kafka SASL 配置

Apache Kafka KRaft 模式配置

Apache Kafka ZooKeeper 模式配置

设置 Apache Kafka 集群

步骤 1：为 Apache Kafka 创建第一个节点

步骤 2：创建第二个节点

步骤3：创建第三个节点

示例：创建一个 replicated topic

All配置

第 1 步：运行 Apache Kafka 映像

第 2 步：编辑配置

维护

备份您的容器

第 1 步：停止当前正在运行的容器

步骤2：运行备份命令

恢复备份

从 Zookeeper 模式迁移到 KRaft 模式

推荐阅读更多精彩内容