目录

• 概述
• 环境准备
• Docker & Docker-Compose
• Linux服务器
• 步骤一：部署到开发环境上
• docker-compose.yml文件编写
• 运行启动脚本
• （拓展）容器可视化页面
• （拓展）Kafka可视化页面
• 用脚本命令进行测试
• 整合Spring Boot应用
• 步骤二：部署到生产环境上
• docker-compose.yml
• docker-compose配置文件变化部分以及说明
• 启动Nginx容器
• 最后一步：IP别名映射的重要性
• extra_hosts配置
• 开发机修改hosts

概述

Kafka的环境配置花了我好几天时间才搞明白，现整理一下。

原来公司对Kafka环境的安装是这样的：

• 获取（下载）Zookeeper和Kafka的Linux安装包
• 安装Zookeeper前保证有Java环境
• 修改配置文件
• 根据官网教程逐步安装

原来的环境部署方式已经有些落后了，现在我们搭建环境基本都会采用Docker容器，不仅简化了部署流程，还方便管理。

在Kafka2.8版本之前，Kafka是强依赖于Zookeeper中间件的，这本身就很不合理，中间件依赖另一个中间件，搭建起来实在麻烦，并且Zookeeper需要搭建三个以上服务作为集群（不考虑挂掉的话一个也可以，但这就不叫集群了，只能算单节点），Kafka也要三个以上做集群（为了数据的高可用），所幸Kafka2.8之后推出了KRaft模式，即抛弃Zookeeper，由Kafka节点自己做Controller来选举Leader，本篇文章内容就是介绍如何在Docker-Compose中搭建Kafka KRaft环境。

环境准备

Docker & Docker-Compose

需要提前准备好Docker和Docker-Compose环境，如果没有安装是不行的

Linux服务器

可以是Windows下的VMware虚拟机，也可以是云服务器

顺带说明一下我当前的环境吧：

• 开发机是Windows，因为需要Linux环境，所以整了个虚拟机安装CentOS 7（有Docker），在这上面搭建我开发时需要的所有环境，比如MySQL、Redis等，这样可以保证开发和生产的环境一致，也不用另一套在Windows上搭建环境的学习成本。
• 虚拟机上的CentOS，上面有说，在这里只安装了Docker & Docker-Compose，然后搭建容器环境给开发时用。
• 云服务器，同样也是CentOS7环境，毕竟光开发是不够的，应用到生产环境上才算部署完整，因为大概率我们会需要Nginx转发等配置，和开发环境会有些许不同，所以最终目的是要在云服务器上搭建完成才可以。

步骤一：部署到开发环境上

首先让我们在开发环境上面部署好Kafka环境，然后写一个Spring Boot应用去连接。

docker-compose.yml文件编写

version: ""
services:
  kafka:
    image: 'bitnami/kafka:.3.1'
    network_mode: mynetwork
    container_name: kafka
    user: root
    ports:
      -:9092
      -:9093
    environment:
      ### 通用配置
      # 允许使用kraft，即Kafka替代Zookeeper
      - KAFKA_ENABLE_KRAFT=yes
      # kafka角色，做broker，也要做controller
      - KAFKA_CFG_PROCESS_ROLES=broker,controller
      # 指定供外部使用的控制类请求信息
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      # 定义kafka服务端socket监听端口
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:,CONTROLLER://:9093
      # 定义安全协议
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      # 使用Kafka时的集群id，集群内的Kafka都要用这个id做初始化，生成一个UUID即可
      - KAFKA_KRAFT_CLUSTER_ID=LelMdIFQkiUFvXCEcqRWA
      # 集群地址
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=@kafka11:9093,2@kafka22:9093,3@kafka33:9093
      # 允许使用PLAINTEXT监听器，默认false，不建议在生产环境使用
      - ALLOW_PLAINTEXT_LISTENER=yes
      # 设置broker最大内存，和初始内存
      - KAFKA_HEAP_OPTS=-XmxM -Xms256M 
      # 不允许自动创建主题
      - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
​
      ### broker配置
      # 定义外网访问地址（宿主机ip地址和端口）
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://.168.1.54:9192
      # broker.id，必须唯一
      - KAFKA_BROKER_ID=
    volumes:
      - /home/mycontainers/kafka/kafka/kraft:/bitnami/kafka
    #extra_hosts:
      #- "kafka:云服务器IP"
      #- "kafka:云服务器IP"
      #- "kafka:云服务器IP"
  kafka:
    image: 'bitnami/kafka:.3.1'
    network_mode: mynetwork
    container_name: kafka
    user: root
    ports:
      -:9092
      -:9093
    environment:
      ### 通用配置
      # 允许使用kraft，即Kafka替代Zookeeper
      - KAFKA_ENABLE_KRAFT=yes
      # kafka角色，做broker，也要做controller
      - KAFKA_CFG_PROCESS_ROLES=broker,controller
      # 指定供外部使用的控制类请求信息
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      # 定义kafka服务端socket监听端口
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:,CONTROLLER://:9093
      # 定义安全协议
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      # 使用Kafka时的集群id，集群内的Kafka都要用这个id做初始化，生成一个UUID即可
      - KAFKA_KRAFT_CLUSTER_ID=LelMdIFQkiUFvXCEcqRWA
      # 集群地址
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=@kafka11:9093,2@kafka22:9093,3@kafka33:9093
      # 允许使用PLAINTEXT监听器，默认false，不建议在生产环境使用
      - ALLOW_PLAINTEXT_LISTENER=yes
      # 设置broker最大内存，和初始内存
      - KAFKA_HEAP_OPTS=-XmxM -Xms256M 
      # 不允许自动创建主题
      - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
​
      ### broker配置
      # 定义外网访问地址（宿主机ip地址和端口）
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://.168.1.54:9292
      # broker.id，必须唯一
      - KAFKA_BROKER_ID=
    volumes:
      - /home/mycontainers/kafka/kafka/kraft:/bitnami/kafka
  kafka:
    image: 'bitnami/kafka:.3.1'
    network_mode: mynetwork
    container_name: kafka
    user: root
    ports:
      -:9092
      -:9093
    environment:
      ### 通用配置
      # 允许使用kraft，即Kafka替代Zookeeper
      - KAFKA_ENABLE_KRAFT=yes
      # kafka角色，做broker，也要做controller
      - KAFKA_CFG_PROCESS_ROLES=broker,controller
      # 指定供外部使用的控制类请求信息
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      # 定义kafka服务端socket监听端口
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:,CONTROLLER://:9093
      # 定义安全协议
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      # 使用Kafka时的集群id，集群内的Kafka都要用这个id做初始化，生成一个UUID即可
      - KAFKA_KRAFT_CLUSTER_ID=LelMdIFQkiUFvXCEcqRWA
      # 集群地址
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=@kafka11:9093,2@kafka22:9093,3@kafka33:9093
      # 允许使用PLAINTEXT监听器，默认false，不建议在生产环境使用
      - ALLOW_PLAINTEXT_LISTENER=yes
      # 设置broker最大内存，和初始内存
      - KAFKA_HEAP_OPTS=-XmxM -Xms256M 
      # 不允许自动创建主题
      - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
​
      ### broker配置
      # 定义外网访问地址（宿主机ip地址和端口）
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://.168.1.54:9392
      # broker.id，必须唯一
      - KAFKA_BROKER_ID=
    volumes:
      - /home/mycontainers/kafka/kafka/kraft:/bitnami/kafka

配置文件说明： 如果你懂一些docker-compose的话，这个配置文件应该很好明白，这里挑一些来说：

镜像选择：bitnami/kafka:3.3.1

这是写这篇文章时最新的版本

容器名是：kafka11、kafka22、kafka33

这个当然随便写，我用11、22、33是为了后面演示生产环境时不冲突，先不用管，后面会说

端口配置是9092和9093：

9092端口用于BROKER传输，即Kafka集群服务端口，我们用Kafka脚本或SpringBoot应用时，连接的就是这个端口；9093是CONTROLLER端口，前面说过，我们抛弃了Zookeeper，用Kafka来代替，这个9093就是充当着原来Zookeeper集群的通讯端口

总结一下，9092用于外网，因为是Kafka要给外部访问；9093用于内网，只用于集群通讯，用于内网是因为环境都搭建在一个服务器的Docker容器内，相当于公司服务器内网，所以不对外开放，生产时一般会搭建在不同的服务器上面，可以是外网也可以是内网，集群的环境配置自由度很高，这点我就不多说明了。

network_mode：

给容器加入网络，这样才可以进行容器间的通讯，可以让容器1中能识别容器2的容器名称，进而解析出IP，很重要，不然只能手动配置容器IP，不过这样很蠢，容器IP不固定，会动态变化。

当然用--link也可以。

KAFKA_KRAFT_CLUSTER_ID：

KRaft模式下，要配置这个集群ID，这个ID可以用Kafka命令去生成：

kafka-storage.sh random-uuid

一开始就用官网示例的LelM2dIFQkiUFvXCEcqRWA就可以了。

KAFKA_BROKER_ID：

broker的id，这个要唯一，搞过Kafka的懂的都懂

KAFKA_CFG_ADVERTISED_LISTENERS：

对外的访问地址，我配置的是192.168.1.54，因为我这台虚拟机用的是桥接网络，分配的IP就是192.168.1.54，这个和容器IP不一样，容器IP只用于容器间的通讯，这个54的配置目的是可以在开发机上（假如开发机的IP是192.168.1.10）通过：192.168.1.54:9092来连接Kafka，很重要

因为有三个Kafka容器，容器内部都用9092和9093端口保持一致，所以端口映射就有所不同：

kafka1配置9192、9193；kafka2配置9292、9293；kafka3配置9392、9393

这个配置也很重要

extra_hosts：

这个配置的目的是修改容器内的hosts文件，增加一个IP的别名映射，在开发阶段用不上，是后面部署生产环境时用的，暂时不用理会，这时候可以先注释掉。

其他的看注释就可以了，还有的是在YML配置文件对Kafka的配置，比如KAFKA_ENABLE_KRAFT，是从Docker Hub的Kafka镜像文档上看的，这是一种快速配置手段，但我们同样可以使用老办法，将kafka的配置文件（server.properties）进行路径映射，如果熟悉docker应该能理解。

运行启动脚本

docker-compose -f docker-compose.yml up

如果没有报错，通过docker ps可以看到Kafka容器：

[root@localhost ~]# docker ps | grep kafkacf69513390   provectuslabs/kafka-ui:latest              "/bin/sh -c 'java $J…"   14 hours ago   Up 14 hours           0.0.0.0:17008->8080/tcp, :::17008->8080/tcp                                                          kafka-ui
cb8c979abb   bitnami/kafka:3.3.1                        "/opt/bitnami/script…"   14 hours ago   Up 4 hours            0.0.0.0:9292->9092/tcp, :::9292->9092/tcp, 0.0.0.0:9293->9093/tcp, :::9293->9093/tcp                 kafka2
f26172ba3e   bitnami/kafka:3.3.1                        "/opt/bitnami/script…"   14 hours ago   Up 4 hours            0.0.0.0:9392->9092/tcp, :::9392->9092/tcp, 0.0.0.0:9393->9093/tcp, :::9393->9093/tcp                 kafka3
e15cd92a   bitnami/kafka:3.3.1                        "/opt/bitnami/script…"   15 hours ago   Up 4 hours            0.0.0.0:9192->9092/tcp, :::9192->9092/tcp, 0.0.0.0:9193->9093/tcp, :::9193->9093/tcp                 kafka1

注意：在启动容器的时候，不出意外的话会出很多意外，检查一下脚本有没有问题，重点排查KAFKA_BROKER_ID和KAFKA_CFG_ADVERTISED_LISTENERS。还有容器名称不要冲突，区分好1、2、3，路径映射也不要弄错，kafka1、kafka2、kafka3。

（拓展）容器可视化页面

如果嫌弃docker ps，可以安装portainer来查看可视化容器页面：

version: ""
services:
  portainer:
    container_name: portainer
    ports:
      -:9000
    restart: always
    network_mode: mynetwork
    volumes:
      - /etc/localtime:/etc/localtime
      - /var/run/docker.sock:/var/run/docker.sock
      - /home/mycontainers/portainer/data:/data:rw
    image: portainer/portainer

访问：http://localhost:9000，即可，请自行修改端口和IP

（拓展）Kafka可视化页面

kafka-ui是Apache出品的Kafka可视化容器，入门是够用了，当然还有很多可视化工具，由于篇幅关系，并且这不是本文重点，所以就不细说了。

version: ""
services:
  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    network_mode: mynetwork
    container_name: kafka-ui
    restart: always
    ports:
      -:8080
    volumes:
      - /etc/localtime:/etc/localtime
    environment:
      # 集群名称
      - KAFKA_CLUSTERS__NAME=local
      # 集群地址
      - KAFKA_CLUSTERS__BOOTSTRAPSERVERS=kafka11:9092,kafka22:9092,kafka33:9092

访问：http://localhost:8080，即可，请自行修改端口和IP

用脚本命令进行测试

Kafka容器内有自带脚本可以使用，所以部署完Kafka环境后，最简单快速的方式就是进入容器内进行测试：

进入任意一个Kafka容器

docker exec -it kafka bash

创建一个主题，名称为demo

kafka-topics.sh --create --topic demo --partitions --replication-factor 3 --bootstrap-server kafka11:9092,kafka22:9092,kafka33:9092

查看所有主题

kafka-topics.sh --bootstrap-server kafka:9092 --list

生产一些消息

kafka-console-producer.sh --bootstrap-server kafka:9092 --topic demo

消费一些消息

kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic demo

制造一些假数据

kafka-producer-perf-test.sh --topic demo --record-size --num-records 1000000 --throughput 10000 --producer-props bootstrap.servers=kafka11:9092 batch.size=16384 linger.ms=0

测试没有问题，命令可以使用，到kafka-ui的可视化页面查看下，有我们刚刚创建的主题，也有我们制造的假数据，三个Kafka节点没有异常，Controller也进行了选举，说明我们的Kafka已经成功创建并运行，还替代了原本Zookeeper的工作，至此，开发环境搭建完成。

整合Spring Boot应用

搭建完了Kafka环境，现在让我们整合到Spring Boot中来使用

• 创建一个Maven项目

配置pom.xml：

<?xml version=".0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/.0.0"
         xmlns:xsi="http://www.w.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>.0.0</modelVersion>
​
    <groupId>com.cc</groupId>
    <artifactId>kafkaDemo</artifactId>
    <version>.0-SNAPSHOT</version>
​
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>.3.4.RELEASE</version>
        <relativePath/>
    </parent>
​
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>.3.4.RELEASE</version>
        </dependency>
    </dependencies>
</project>

新建应用程序启动类：

@SpringBootApplication
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

配置application.yml，去连接Kafka环境

server:
  port:
​
spring:
  application:
    name: kafkaDemo
​
  kafka:
    bootstrap-servers: mylocalhost:,mylocalhost:9292,mylocalhost:9392
    producer:
      # 生产者序列化
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      # ack策略
      #：生产者发送消息就不管了，效率高，但是容易丢数据，且没有重试机制
      #：消息发送到Leader并落盘后就返回，如果Leader挂了并且Follower还没有同步数据就会丢失数据
      # -：消息要所有副本都罗盘才返回，保证数据不丢失（但是有可能重复消费）
      acks: -
      # 失败重试次数
      retries:
      # 批量提交的数据大小
      batch-size:
      # 生产者暂存数据的缓冲区大小
      buffer-memory:
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      # 是否自动提交偏移量，如果要手动确认消息，就要设置为false
      enable-auto-commit: false
      # 消费消息后间隔多长时间提交偏移量（ms）
      auto-commit-interval:
      # 默认的消费者组，如果不指定就会用这个
      group-id: mykafka
      # kafka意外宕机时的消息消费策略
      # earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
      # latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
      # none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
      auto-offset-reset: latest
    listener:
      # 手动确认消息
      ack-mode: manual_immediate
      # 消费者运行的线程数
      concurrency:

我们只测试连接，不演示消息的生产和消费，现在关注application.yml配置文件，我们只看这个配置：

bootstrap-servers: mylocalhost:,mylocalhost:9292,mylocalhost:9392

mylocalhost是我在本机hosts中设置的IP别名映射，即：mylocalhost=192.168.1.54

9192、9292、9392是kafka集群容器映射 出来对外的端口，查看一下docker-compose.yml文件即可明白。

启动程序，应用连接Kafka，出现了...JVM...字样，表示成功连接Kafka。

步骤二：部署到生产环境上

生产环境和开发环境有什么不同呢？不同在我们往往没有那么多的IP资源可以对外开放，则需要做Nginx转发。

比如说Kafka集群是在内网机器中部署，最终由一台代理机器转发Kafka集群，当然这个代理机器实际上也是要做负载均衡的，不然代理机器挂了整个集群就无了，不过Nginx的负载均衡不在本文篇幅中。

先来看看生产环境的docker-compose.yml：

docker-compose.yml

version: ""
services:
  kafka:
    image: 'bitnami/kafka:.3.1'
    network_mode: mynetwork
    container_name: kafka
    user: root
    environment:
      ### 通用配置
      # 允许使用kraft，即Kafka替代Zookeeper
      - KAFKA_ENABLE_KRAFT=yes
      # kafka角色，做broker，也要做controller
      - KAFKA_CFG_PROCESS_ROLES=broker,controller
      # 指定供外部使用的控制类请求信息
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      # 定义kafka服务端socket监听端口
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:,CONTROLLER://:9093
      # 定义安全协议
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      # 集群地址
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=@kafka1:9093,2@kafka2:9093,3@kafka3:9093
      # 允许使用PLAINTEXT监听器，默认false，不建议在生产环境使用
      - ALLOW_PLAINTEXT_LISTENER=yes
      # 设置broker最大内存，和初始内存
      - KAFKA_HEAP_OPTS=-XmxM -Xms256M 
      # 不允许自动创建主题
      - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
      # 使用Kafka时的集群id，集群内的Kafka都要用这个id做初始化，生成一个UUID即可
      - KAFKA_KRAFT_CLUSTER_ID=LelMdIFQkiUFvXCEcqRWA
​
      ### broker配置
      # 定义外网访问地址（宿主机ip地址和端口）
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:17005
      # broker.id，必须唯一
      - KAFKA_BROKER_ID=
    volumes:
      - /home/mycontainers/kafka/kafka/kraft:/bitnami/kafka
  kafka:
    image: 'bitnami/kafka:.3.1'
    network_mode: mynetwork
    container_name: kafka
    user: root
    environment:
      ### 通用配置
      # 允许使用kraft，即Kafka替代Zookeeper
      - KAFKA_ENABLE_KRAFT=yes
      # kafka角色，做broker，也要做controller
      - KAFKA_CFG_PROCESS_ROLES=broker,controller
      # 指定供外部使用的控制类请求信息
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      # 定义kafka服务端socket监听端口
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:,CONTROLLER://:9093
      # 定义安全协议
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      # 集群地址
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=@kafka1:9093,2@kafka2:9093,3@kafka3:9093
      # 允许使用PLAINTEXT监听器，默认false，不建议在生产环境使用
      - ALLOW_PLAINTEXT_LISTENER=yes
      # 设置broker最大内存，和初始内存
      - KAFKA_HEAP_OPTS=-XmxM -Xms256M 
      # 不允许自动创建主题
      - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
      # 使用Kafka时的集群id，集群内的Kafka都要用这个id做初始化，生成一个UUID即可
      - KAFKA_KRAFT_CLUSTER_ID=LelMdIFQkiUFvXCEcqRWA
​
      ### broker配置
      # 定义外网访问地址（宿主机ip地址和端口）
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:17005
      # broker.id，必须唯一
      - KAFKA_BROKER_ID=
​
    volumes:
      - /home/mycontainers/kafka/kafka/kraft:/bitnami/kafka
  kafka:
    image: 'bitnami/kafka:.3.1'
    network_mode: mynetwork
    container_name: kafka
    user: root
    environment:
      ### 通用配置
      # 允许使用kraft，即Kafka替代Zookeeper
      - KAFKA_ENABLE_KRAFT=yes
      # kafka角色，做broker，也要做controller
      - KAFKA_CFG_PROCESS_ROLES=broker,controller
      # 指定供外部使用的控制类请求信息
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      # 定义kafka服务端socket监听端口
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:,CONTROLLER://:9093
      # 定义安全协议
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      # 集群地址
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=@kafka1:9093,2@kafka2:9093,3@kafka3:9093
      # 允许使用PLAINTEXT监听器，默认false，不建议在生产环境使用
      - ALLOW_PLAINTEXT_LISTENER=yes
      # 设置broker最大内存，和初始内存
      - KAFKA_HEAP_OPTS=-XmxM -Xms256M 
      # 不允许自动创建主题
      - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
      # 使用Kafka时的集群id，集群内的Kafka都要用这个id做初始化，生成一个UUID即可
      - KAFKA_KRAFT_CLUSTER_ID=LelMdIFQkiUFvXCEcqRWA
​
      ### broker配置
      # 定义外网访问地址（宿主机ip地址和端口）
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:17005
      # broker.id，必须唯一
      - KAFKA_BROKER_ID=
    volumes:
      - /home/mycontainers/kafka/kafka/kraft:/bitnami/kafka

注意，这里我们的容器名是kafka1、kafka2、kafka3，和开发环境有所区别。

接下来是重点部分，要理解重点配置才可以正常使用，生产环境这部分我花了两天才搞通。

docker-compose配置文件变化部分以及说明

先重新声明一些题要：

• 开发机的IP是：192.168.1.10
• 开发环境的虚拟机IP是：192.168.1.54（与开发机同网络）
• Kafka集群处于内网环境，比如公司的内部服务器，无法对外访问
• 云服务器的IP是：a.a.a.a（自行脑补）,云服务器就是代理服务器，可以访问Kafka集群服务
• 开发环境的Kafka容器名是：kafka11、kafka22、kafka33
• 生产环境的Kafka容器名是：kafka1、kafka2、kafka3

接下来对生产环境的配置变动进行说明：

• 去掉端口映射，因为我们等会会创建一个Ngnix容器来转发，容器间已经可以进行通讯，所以就不需要对外端口了，除非这个Nginx容器是另外一台机器上，那么就需要对外端口。
• KAFKA_CFG_ADVERTISED_LISTENERS的外网访问地址从实际IP改成了kafka1:17005， 在开发环境中，我们的开发机子可以通过虚拟机IP来访问容器，所以配置192.168.1.54，但是生产环境这里我们不能直接访问了（假设生产环境的Kafka集群是在内网），我们只能访问代理服务器，让代理服务器帮忙转发请求，所以这里改的：kafka1:17005，必须是代理服务器可以访问到的，因为我们代理服务器和生产环境的Kafka集群是同一个容器组内，所以可以访问，这是为了便于演示，实际上代理服务器和Kafka集群肯定不会在同一台机器内，所以就不能用：kafka1:17005，而是要用：[代理服务器可以访问到的Kafka集群地址]:17005
• 17005端口替换原来的9092端口，因为我的云服务器安全组没有开放9092，所以改成17005，这个端口要和等会转发用的Nginx端口保持一致，即Nginx容器也要开放17005端口

启动Nginx容器

nginx.yml：

version: ""
services:
  nginx:
    image: nginx:latest
    network_mode: mynetwork
    container_name: nginx
    restart: always
    ports:
      -:9092
    volumes:
      - /etc/localtime:/etc/localtime
      - /home/mycontainers/nginx/nginx.conf:/etc/nginx/nginx.conf
      - /home/mycontainers/nginx/logs:/var/log/nginx
      - /home/mycontainers/nginx/conf.d/:/etc/nginx/conf.d

手动修改nginx.conf：

stream {
    upstream kafka {
        server kafka:17005;
        server kafka:17005;
        server kafka:17005;
    }
​
    server {
        listen;
        proxy_pass kafka;
    }
}
​
http {
    ... 
    location / {
        ...
    }
    ...
}

启动Nginx容器后，修改一下配置，将Kaka集群配置进来，然后监听9092端口进行转发，当然端口可以定制，这个不重要，重要的是Nginx容器对外的17005端口，流程是这样的：

• 外部访问17005端口，映射到Nginx的9092端口
• Nginx的9092端口对应了Kafka的集群，Kafka集群的端口是17005，所以Nginx的对外也要是17005，这样要保证强一致

至此，生产环境搭建完成。

最后一步：IP别名映射的重要性

还记得上面搭建开发环境时候先注释掉的extra_hosts配置吗：

# 刚刚我们有进行声明，云服务器IP是a.a.a.a，所以去掉注释，自行替换
extra_hosts:
    - "kafka:云服务器IP"
    - "kafka:云服务器IP"
    - "kafka:云服务器IP"

现在用上了，这个只配置到了开发环境的kafka11容器中，kafka22是没有的，我们先进入kafka22容器中去连接生产环境的Kafka集群看看：

进入kafka22容器

docker exec -it kafka bash

连接生产环境的Kafka集群（通过Nginx转发）

kafka-topics.sh --bootstrap-server a.a.a.a: --list

会报错：

[2023-01-11 07:29:07,495] WARN [AdminClient clientId=adminclient-1] Error connecting to node kafka3:17005 (id: 3 rack: null) (org.apache.kafka.clients.NetworkClient)

java.net.UnknownHostException: kafka3

  at java.base/java.net.InetAddress$CachedAddresses.get(InetAddress.java:797)

  at java.base/java.net.InetAddress.getAllByName0(InetAddress.java:1519)

  at java.base/java.net.InetAddress.getAllByName(InetAddress.java:1378)

  at java.base/java.net.InetAddress.getAllByName(InetAddress.java:1306)

  at org.apache.kafka.clients.DefaultHostResolver.resolve(DefaultHostResolver.java:27)

  at org.apache.kafka.clients.ClientUtils.resolve(ClientUtils.java:110)

  at org.apache.kafka.clients.ClusterConnectionStates$NodeConnectionState.currentAddress(ClusterConnectionStates.java:510)

  at org.apache.kafka.clients.ClusterConnectionStates$NodeConnectionState.access$200(ClusterConnectionStates.java:467)

  at org.apache.kafka.clients.ClusterConnectionStates.currentAddress(ClusterConnectionStates.java:173)

  at org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:990)

  at org.apache.kafka.clients.NetworkClient.ready(NetworkClient.java:301)

  at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.sendEligibleCalls(KafkaAdminClient.java:1143)

  at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.processRequests(KafkaAdminClient.java:1403)

  at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1346)

  at java.base/java.lang.Thread.run(Thread.java:829)

可以看到关键字：kafka3:17005，在开发环境中连接生产环境报这个kafka3，而kafka3是在生产环境中配置的，我们开发环境配置的是kafka33，所以很明显，在连接Kafka的时候，会自动去读取集群地址，就是我们生产环境的docker-compose.yml中的：

  - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=@kafka1:9093,2@kafka2:9093,3@kafka3:9093

虽然Kafka集群运行起来了，Nginx转发成功了，但是实际连接不上，因为不知道这个kafka3是什么意思，所以我们还需要做最后一步：IP别名映射

extra_hosts配置

退出kafka22容器，恢复kafka11容器的extra_hosts配置：

extra_hosts:
    - "kafka:a.a.a.a"
    - "kafka:a.a.a.a"
    - "kafka:a.a.a.a"

重新启动开发环境的Kafka容器，然后进入kafka11容器去连接生产环境，这时候就成功了，因为会把kafka1、kafka2、kafka3都映射到a.a.a.a即代理服务器的公网IP，所以至此，生产环境搭建完成。

开发机修改hosts

我们现在知道，连接生产环境的Kafka获取的集群地址是kafka1、kafka2、kafka3，所以在开发机中我们同样需要修改hosts配置，映射实际的公网IP，不然无法识别。