Kafka穿过网闸（物理层）进行消费

刺骨的言语ヽ痛彻心扉 2022-12-29 06:50 119阅读 0赞

### Kafka穿过网闸（物理层）进行消费 ###

*  前言
 *  服务器信息
 *   *  互联网服务器A：
     *  网闸机器B：
     *  内网服务器C：
 *  Kafka server.properties配置
 *  测试内网消费者（Springboot）
 *   *  代码
     *  application.properties
 *  原理
 *   *  新消费组加入的过程

# 前言 #

当前公司有一个场景，就是在互联网服务器搭建kafka集群，在内网环境下进行互联网Kafka集群的消费，在内外网之间搭建一层物理层面的网闸来代理外网服务器。  
  由于对安全的考虑，内外网服务器都打开了防火墙。当时只使用互联网Kafka集群对内网机器开通了`9092`、`2181`端口。但是不管怎么对集群进行配置都都达不到穿网消费的效果，消费者始终TimeOut。  
  最终又去研究了下Kafka消费时网络请求机制才发现，单项端口开通并达不到消费效果，必须双向开通端口才行。

# 服务器信息 #

这里用三台机器举例，互联网、内网、网闸各一台

## 互联网服务器A： ##

# 主机名
    A
    # IP
    172.17.xxx.xx
    
    # 主机名及IP映射
    172.17.xxx.xx A

## 网闸机器B： ##

机器属于购买第三方物理网闸，网闸B代理互联网服务器A，只需要在这台机器开放对A的防火墙限制即可。

# ip
    10.1.xx.xx

## 内网服务器C： ##

# 主机名
    C
    # IP
    192.168.xxx.xx
    
    # 主机名及IP映射
    192.168.xxx.xx C
    10.1.xx.xx	   A

# Kafka server.properties配置 #

其他配置保持正常修改下面参数IP改为Kafka所在服务器主机名

advertised.listeners=PLAINTEXT://A:9092

# 测试内网消费者（Springboot） #

## 代码 ##

@SpringBootApplication
    public class Kafka { 
    	public static void main(String[] args) { 
    		SpringApplication.run(Kafka.class, args);
    	}
    
    	@KafkaListener(topics = "test_topic", groupId = "test")
    	public void kafkaConsumerListener(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) { 
    		System.out.println(record);
    	}
    }

## application.properties ##

spring.kafka.consumer.bootstrap-servers = A:9092
    spring.kafka.consumer.enable-auto-commit=false
    spring.kafka.listener.ack-mode=manual
    spring.kafka.consumer.max-poll-records=100
    spring.kafka.consumer.auto-offset-reset=latest
    
    spring.kafka.listener.concurrency=1
    spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
    spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

将项目打包放到内网机器执行，可以完整地消费到数据，穿网消费成功。

# 原理 #

众所周知Kafka消费时是根据消费组来进行消费的，当有一个新的消费组加入时，会触发Kafka的Rebalance机制。

## 新消费组加入的过程 ##

新的消费组第一次请求消费时会有以下交互（仅列出新组员加入时的请求与响应）：

# 确认网络畅通
    1. 新的Group向Coordinator发送心跳请求
    2. Container响应Group
    
    # 开启Rebalance
    3. 向Coordinator申请开启rebalance
    4. 向group开启rebalance，并让group重新加入
    
    # 注册消费者信息
    5. 请求加入消费组
    6. Coordinator注册消费者信息

由上述步骤可看出，Kafka拉取一次数据是要进行多次双向交互，所以要在物理网闸上面开`双向端口`。