概述
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: "3"
services:
kafka1:
image: 'bitnami/kafka:3.3.1'
network_mode: mynetwork
container_name: kafka11
user: root
ports:
- 9192:9092
- 9193: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://:9092,CONTROLLER://:9093
# 定义安全协议
- KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
# 使用Kafka时的集群id,集群内的Kafka都要用这个id做初始化,生成一个UUID即可
- KAFKA_KRAFT_CLUSTER_ID=LelM2dIFQkiUFvXCEcqRWA
# 集群地址
- KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka11:9093,2@kafka22:9093,3@kafka33:9093
# 允许使用PLAINTEXT监听器,默认false,不建议在生产环境使用
- ALLOW_PLAINTEXT_LISTENER=yes
# 设置broker最大内存,和初始内存
- KAFKA_HEAP_OPTS=-Xmx512M -Xms256M
# 不允许自动创建主题
- KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
### broker配置
# 定义外网访问地址(宿主机ip地址和端口)
- KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://192.168.1.54:9192
# broker.id,必须唯一
- KAFKA_BROKER_ID=1
volumes:
- /home/mycontainers/kafka1/kafka/kraft:/bitnami/kafka
#extra_hosts:
#- "kafka1:云服务器IP"
#- "kafka2:云服务器IP"
#- "kafka3:云服务器IP"
kafka2:
image: 'bitnami/kafka:3.3.1'
network_mode: mynetwork
container_name: kafka22
user: root
ports:
- 9292:9092
- 9293: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://:9092,CONTROLLER://:9093
# 定义安全协议
- KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
# 使用Kafka时的集群id,集群内的Kafka都要用这个id做初始化,生成一个UUID即可
- KAFKA_KRAFT_CLUSTER_ID=LelM2dIFQkiUFvXCEcqRWA
# 集群地址
- KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka11:9093,2@kafka22:9093,3@kafka33:9093
# 允许使用PLAINTEXT监听器,默认false,不建议在生产环境使用
- ALLOW_PLAINTEXT_LISTENER=yes
# 设置broker最大内存,和初始内存
- KAFKA_HEAP_OPTS=-Xmx512M -Xms256M
# 不允许自动创建主题
- KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
### broker配置
# 定义外网访问地址(宿主机ip地址和端口)
- KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://192.168.1.54:9292
# broker.id,必须唯一
- KAFKA_BROKER_ID=2
volumes:
- /home/mycontainers/kafka2/kafka/kraft:/bitnami/kafka
kafka3:
image: 'bitnami/kafka:3.3.1'
network_mode: mynetwork
container_name: kafka33
user: root
ports:
- 9392:9092
- 9393: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://:9092,CONTROLLER://:9093
# 定义安全协议
- KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
# 使用Kafka时的集群id,集群内的Kafka都要用这个id做初始化,生成一个UUID即可
- KAFKA_KRAFT_CLUSTER_ID=LelM2dIFQkiUFvXCEcqRWA
# 集群地址
- KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka11:9093,2@kafka22:9093,3@kafka33:9093
# 允许使用PLAINTEXT监听器,默认false,不建议在生产环境使用
- ALLOW_PLAINTEXT_LISTENER=yes
# 设置broker最大内存,和初始内存
- KAFKA_HEAP_OPTS=-Xmx512M -Xms256M
# 不允许自动创建主题
- KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
### broker配置
# 定义外网访问地址(宿主机ip地址和端口)
- KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://192.168.1.54:9392
# broker.id,必须唯一
- KAFKA_BROKER_ID=3
volumes:
- /home/mycontainers/kafka3/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 kafka
80cf69513390 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
c66b8c979abb 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
70f26172ba3e 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
0193e15cd92a 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: "3"
services:
portainer:
container_name: portainer
ports:
- 9000: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: "3"
services:
kafka-ui:
image: provectuslabs/kafka-ui:latest
network_mode: mynetwork
container_name: kafka-ui
restart: always
ports:
- 8080:8080
volumes:
- /etc/localtime:/etc/localtime
environment:
# 集群名称
- KAFKA_CLUSTERS_0_NAME=local
# 集群地址
- KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=kafka11:9092,kafka22:9092,kafka33:9092
访问:http://localhost:8080,即可,请自行修改端口和IP
用脚本命令进行测试
Kafka容器内有自带脚本可以使用,所以部署完Kafka环境后,最简单快速的方式就是进入容器内进行测试:
进入任意一个Kafka容器
docker exec -it kafka11 bash
创建一个主题,名称为demo
kafka-topics.sh --create --topic demo --partitions 3 --replication-factor 3 --bootstrap-server kafka11:9092,kafka22:9092,kafka33:9092
查看所有主题
kafka-topics.sh --bootstrap-server kafka11:9092 --list
生产一些消息
kafka-console-producer.sh --bootstrap-server kafka11:9092 --topic demo
消费一些消息
kafka-console-consumer.sh --bootstrap-server kafka11:9092 --topic demo
制造一些假数据
kafka-producer-perf-test.sh --topic demo --record-size 1024 --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="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.cc</groupId>
<artifactId>kafkaDemo</artifactId>
<version>1.0-SNAPSHOT</version>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.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>2.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: 8888
spring:
application:
name: kafkaDemo
kafka:
bootstrap-servers: mylocalhost:9192,mylocalhost:9292,mylocalhost:9392
producer:
# 生产者序列化
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
# ack策略
# 0:生产者发送消息就不管了,效率高,但是容易丢数据,且没有重试机制
# 1:消息发送到Leader并落盘后就返回,如果Leader挂了并且Follower还没有同步数据就会丢失数据
# -1:消息要所有副本都罗盘才返回,保证数据不丢失(但是有可能重复消费)
acks: -1
# 失败重试次数
retries: 3
# 批量提交的数据大小
batch-size: 16384
# 生产者暂存数据的缓冲区大小
buffer-memory: 33554432
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: 100
# 默认的消费者组,如果不指定就会用这个
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: 2
我们只测试连接,不演示消息的生产和消费,现在关注application.yml配置文件,我们只看这个配置:
bootstrap-servers: mylocalhost:9192,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: "3"
services:
kafka1:
image: 'bitnami/kafka:3.3.1'
network_mode: mynetwork
container_name: kafka1
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://:17005,CONTROLLER://:9093
# 定义安全协议
- KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
# 集群地址
- KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka1:9093,2@kafka2:9093,3@kafka3:9093
# 允许使用PLAINTEXT监听器,默认false,不建议在生产环境使用
- ALLOW_PLAINTEXT_LISTENER=yes
# 设置broker最大内存,和初始内存
- KAFKA_HEAP_OPTS=-Xmx512M -Xms256M
# 不允许自动创建主题
- KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
# 使用Kafka时的集群id,集群内的Kafka都要用这个id做初始化,生成一个UUID即可
- KAFKA_KRAFT_CLUSTER_ID=LelM2dIFQkiUFvXCEcqRWA
### broker配置
# 定义外网访问地址(宿主机ip地址和端口)
- KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka1:17005
# broker.id,必须唯一
- KAFKA_BROKER_ID=1
volumes:
- /home/mycontainers/kafka1/kafka/kraft:/bitnami/kafka
kafka2:
image: 'bitnami/kafka:3.3.1'
network_mode: mynetwork
container_name: kafka2
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://:17005,CONTROLLER://:9093
# 定义安全协议
- KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
# 集群地址
- KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka1:9093,2@kafka2:9093,3@kafka3:9093
# 允许使用PLAINTEXT监听器,默认false,不建议在生产环境使用
- ALLOW_PLAINTEXT_LISTENER=yes
# 设置broker最大内存,和初始内存
- KAFKA_HEAP_OPTS=-Xmx512M -Xms256M
# 不允许自动创建主题
- KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
# 使用Kafka时的集群id,集群内的Kafka都要用这个id做初始化,生成一个UUID即可
- KAFKA_KRAFT_CLUSTER_ID=LelM2dIFQkiUFvXCEcqRWA
### broker配置
# 定义外网访问地址(宿主机ip地址和端口)
- KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka2:17005
# broker.id,必须唯一
- KAFKA_BROKER_ID=2
volumes:
- /home/mycontainers/kafka2/kafka/kraft:/bitnami/kafka
kafka3:
image: 'bitnami/kafka:3.3.1'
network_mode: mynetwork
container_name: kafka3
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://:17005,CONTROLLER://:9093
# 定义安全协议
- KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
# 集群地址
- KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka1:9093,2@kafka2:9093,3@kafka3:9093
# 允许使用PLAINTEXT监听器,默认false,不建议在生产环境使用
- ALLOW_PLAINTEXT_LISTENER=yes
# 设置broker最大内存,和初始内存
- KAFKA_HEAP_OPTS=-Xmx512M -Xms256M
# 不允许自动创建主题
- KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
# 使用Kafka时的集群id,集群内的Kafka都要用这个id做初始化,生成一个UUID即可
- KAFKA_KRAFT_CLUSTER_ID=LelM2dIFQkiUFvXCEcqRWA
### broker配置
# 定义外网访问地址(宿主机ip地址和端口)
- KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka3:17005
# broker.id,必须唯一
- KAFKA_BROKER_ID=3
volumes:
- /home/mycontainers/kafka3/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: "3"
services:
nginx:
image: nginx:latest
network_mode: mynetwork
container_name: nginx
restart: always
ports:
- 17005: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 kafka1:17005;
server kafka2:17005;
server kafka3:17005;
}
server {
listen 9092;
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:
- "kafka1:云服务器IP"
- "kafka2:云服务器IP"
- "kafka3:云服务器IP"
现在用上了,这个只配置到了开发环境的kafka11容器中,kafka22是没有的,我们先进入kafka22容器中去连接生产环境的Kafka集群看看:
进入kafka22容器
docker exec -it kafka22 bash
连接生产环境的Kafka集群(通过Nginx转发)
kafka-topics.sh --bootstrap-server a.a.a.a:17005 --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=1@kafka1:9093,2@kafka2:9093,3@kafka3:9093
虽然Kafka集群运行起来了,Nginx转发成功了,但是实际连接不上,因为不知道这个kafka3是什么意思,所以我们还需要做最后一步:IP别名映射
extra_hosts配置
退出kafka22容器,恢复kafka11容器的extra_hosts配置:
extra_hosts:
- "kafka1:a.a.a.a"
- "kafka2:a.a.a.a"
- "kafka3:a.a.a.a"
重新启动开发环境的Kafka容器,然后进入kafka11容器去连接生产环境,这时候就成功了,因为会把kafka1、kafka2、kafka3都映射到a.a.a.a即代理服务器的公网IP,所以至此,生产环境搭建完成。
开发机修改hosts
我们现在知道,连接生产环境的Kafka获取的集群地址是kafka1、kafka2、kafka3,所以在开发机中我们同样需要修改hosts配置,映射实际的公网IP,不然无法识别。
以上就是详细讲解Docker-Compose部署Kafka KRaft集群环境的详细内容,更多关于Docker Compose部署Kafka KRaft的资料请关注编程网其它相关文章!