背景说明
我们在使用Kafka客户端连接到Kafka集群时，即使连接的节点只配置了一个集群的Broker地址，该Broker将返回给客户端集群所有节点的信息列表。然后客户端使用该列表信息（Topic的分区信息）再与集群进行数据交互。这里Kafka列表信息为服务配置文件
service.properties中advertised.listeners配置项中的信息。例如：advertised.listeners=PLAINTEXT://192.168.1.1:9092这样在通信中就存在一个网络连通性问题。如果Kafka位于内网环境，而客户端位于外网环境，即使内外网配置了IP地址映射（网络层面的NAT），由于返回给外网客户端的IP列表是内网地址，客户端和Broker第一次通讯获取集群元数据中包含是
advertised.listeners配置中的内网地址信息，由于内外网隔离，就会出现客户端和集群的通信无法通讯的报错。通常对于这种情况的解决方案是Kafka集群的
advertised.listeners配置项使用主机名方式。例如：advertised.listeners=PLAINTEXT://Kafka.node1:9092
在实际业务场景中，系统架构上确实存在需要使用Nginx的场景，这时候就需要我们在架构设计上要符合Kafka的通讯机制。
kafka集群服务端
1. 配置kafka集群
# kafka1 节点1cat server.properties |grep -v ^# | grep -v ^$
broker.id=0
listeners=INTERNAL://192.168.100.100:9093,EXTERNAL://kafka1:9092
inter.broker.listener.name=INTERNAL
advertised.listeners=INTERNAL://192.168.100.100:9093,EXTERNAL://kafka1:9092
listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/kafka/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.100.100:2181,192.168.100.101:2181,192.168.100.102:2181
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0# kafka2 节点2cat server.properties |grep -v ^# | grep -v ^$
broker.id=0
listeners=INTERNAL://192.168.100.101:9093,EXTERNAL://kafka2:9092
inter.broker.listener.name=INTERNAL
advertised.listeners=INTERNAL://192.168.100.101:9093,EXTERNAL://kafka2:9092
listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/kafka/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.100.100:2181,192.168.100.101:2181,192.168.100.102:2181
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0# kafka3 节点3cat server.properties |grep -v ^# | grep -v ^$
broker.id=0
listeners=INTERNAL://192.168.100.102:9093,EXTERNAL://kafka3:9092
inter.broker.listener.name=INTERNAL
advertised.listeners=INTERNAL://192.168.100.102:9093,EXTERNAL://kafka3:9092
listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/kafka/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.100.100:2181,192.168.100.101:2181,192.168.100.102:2181
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=02. 配置hosts映射
# 三个节点均需要配置主机名映射cat /etc/hosts192.168.100.100 kafka1
192.168.100.101 kafka2
192.168.100.102 kafka33. 启动命令示例
./bin/kafka-server-start.sh -daemon config/server.properties
Nginx中间代理
nginx代理作为kafka服务端和客户端隔离的中转，因此需要两边的网络都能访问。
1. 配置hosts主机名映射
192.168.100.100 kafka1
192.168.100.101 kafka2
192.168.100.102 kafka32. 配置TCP四层代理
nginx四层代理需要使用stream模块，因此nginx一般建议使用rpm包安装或编译安装开启此模块。
# nginx 实例1
stream {server{listen 9092;proxy_pass node1;}upstream node1{server kafka1:9092 weight=1;}
}# nginx实例2stream {server{listen 9092;proxy_pass node2;}upstream node2{server kafka2:9092 weight=1;}
}# nginx实例3stream {#1server{listen 9092;proxy_pass node3;}upstream node3{server kafka3:9092 weight=1;}
}kafka客户端
1. 配置hosts主机名映射
cat /etc/hosts192.168.200.100 kafka1
192.168.200.101 kafka2
192.168.200.102 kafka3192.168.200.x 为nginx节点所在ip2. 访问测试验证
# 查看有哪些topickafka-topics.sh --bootstrap-server kafka1:9092 --list# 创建一个topickafka-topics.sh --bootstrap-server kafka1:9092 --create --replication-factor 2 --partitions 30 --topic test-topic# 生产一条消息echo '{"key":"value"}' | kafka-console-producer.sh --broker-list kafka1:9092 --topic test-topic# 消费数据kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic test-topic --from-beginning --max-messages 3架构说明
数据流说明
（1）客户端通过Nginx代理获取到Kafka集群的元数据信息。
（2）元数据信息中包含的是集群的主机名信息，客户端获取到主机名后，在hosts文件中进行解析。例如：kafka1映射为对应的Nginx实例地址：192.168.200.100。这样数据流就会再次Nginx进行交互，相当于对Kafka客户端进行了"欺骗"
总结
使用Nginx代理Kafka集群，架构并没有较少对外暴露服务的实例数量。架构上主要能实现内外网隔离安全。例如外网Kafka客户端不用开通到内网Kafka集群的直连火墙。只需要将Nginx集群的地址暴露给外网客户端。
架构上可以将Kafka集群中三个节点的监听端口分别配置成不同端口，例如：
advertised.listeners=PLAINTEXT://kafka1:9092
advertised.listeners=PLAINTEXT://kafka2:9093
advertised.listeners=PLAINTEXT://kafka3:9094在一台节点服务器上运行一个Nginx，然后配置三个不同的server分别监听不同的端口:
stream {#1server{listen 9092;proxy_pass node1;}upstream node1{server kafka1:9092 weight=1;}#2server{listen 9093;proxy_pass node2;}upstream node2{server kafka2:9093 weight=1;}#3server{listen 9094;proxy_pass node3;}upstream node3{server kafka3:9094 weight=1;}
}在资源上我们只需要部署一台对外暴露的Nginx服务器IP和三个端口即可
192.168.200.100 kafka1
192.168.200.100 kafka2
192.168.200.100 kafka3这样客户端就避免和Kafka Broker直连，而是通过nginx进行了路由。
