Kafka 管理TCP连接

生产者管理TCP连接

Kafka生产者程序概览

Kafka的Java生产者API主要的对象就是KafkaProducer。通常我们开发一个生产者的步骤有4步：

第1步：构造生产者对象所需的参数对象。

第2步：利用第1步的参数对象，创建KafkaProducer对象实例。

第3步：使用KafkaProducer的send方法发送消息。

第4步：调用KafkaProducer的close方法关闭生产者并释放各种系统资源。

上面这4步写成Java代码的话大概是这个样子：

Properties props = new Properties ();
props.put(“参数1”, “参数1的值”)；
props.put(“参数2”, “参数2的值”)；
……
try (Producer<String, String> producer = new KafkaProducer<>(props)) {producer.send(new ProducerRecord<String, String>(……), callback);……
}

何时创建TCP连接

1. TCP连接在创建KafkaProducer实例时建立

在创建KafkaProducer实例时，生产者应用会在后台创建并启动一个名为 Sender的线程，该Sender线程开始运行时首先会创建与Broker的TCP连接的。

如果不调用send方法，这个Producer都不知道给哪个主题发消息，它又怎么能知道连接哪个Broker呢？这是通过Producer的核心参数之一bootstrap.servers参数来指定的。

如果为这个参数指定了1000个Broker连接信息，Producer启动时会首先创建与这1000个Broker的TCP连接，所以不建议把集群中所有的Broker信息都配置到bootstrap.servers中，因为Producer一旦连接到集群中的任一台Broker，就能拿到整个集群的Broker信息，故没必要为bootstrap.servers指定所有的Broker。通过日志可以看出：

[2018-12-09 09:35:45,828] DEBUG[ProducerclientId=producer-1] Sendingmetadatarequest (type=MetadataRequest, topics=) to nodelocalhost:9093 (id:-2 rack: null) (org.apache.kafka.clients.NetworkClient:1068)

Producer向某一台Broker发送了MetadataRequest请求，尝试获取集群的元数据信息——这就是前面提到的Producer能够获取集群所有信息的方法。

2. 在更新元数据后和在消息发送时

(1) 当Producer尝试给一个不存在的主题发送消息时，Broker会告诉Producer说这个主题不存在。此时Producer会发送METADATA请求给Kafka集群，去尝试获取最新的元数据信息。

(2) Producer通过metadata.max.age.ms参数定期地去更新元数据信息。该参数的默认值是300000，即5分钟，也就是说不管集群那边是否有变化，Producer每5分钟都会强制刷新一次元数据以保证它是最及时的数据。

何时关闭TCP连接

1. 用户主动关闭

这里的主动关闭实际上是广义的主动关闭，甚至包括用户调用kill-9主动“杀掉”Producer应用。当然最推荐的方式还是调用producer.close()方法来关闭。

2. Kafka自动关闭

这与Producer端参数connections.max.idle.ms的值有关。默认情况下该参数值是9分钟，即如果在9分钟内没有任何请求“流过”某个TCP连接，那么Kafka会主动帮你把该TCP连接关闭。用户可以在Producer端设置connections.max.idle.ms=-1禁掉这种机制。一旦被设置成-1，TCP连接将成为永久长连接。当然这只是软件层面的“长连接”机制，由于Kafka创建的这些Socket连接都开启了 keepalive，因此keepalive探活机制还是会遵守的。

自动关闭中，TCP连接是在Broker端被关闭的，但其实这个TCP连接的发起方是客户端，因此在TCP看来，这属于被动关闭的场景，即passive close。被动关闭的后果就是会产生大量的 CLOSE_WAIT连接，因此Producer端或Client端没有机会显式地观测到此连接已被中断。

小结：

Java Producer端管理TCP连接的方式是：

1. KafkaProducer实例创建时启动Sender线程，从而创建与bootstrap.servers中所有Broker的TCP连接。

2. KafkaProducer实例首次更新元数据信息之后，还会再次创建与集群中所有Broker的TCP连接。

3. 如果Producer端发送消息到某台Broker时发现没有与该Broker的TCP连接，那么也会立即创建连接。

4. 如果设置Producer端connections.max.idle.ms参数大于0，则步骤1中创建的TCP连接会被自动关闭；如果设置该参数=-1，那么步骤1中创建的TCP连接将无法被关闭，从而成为“僵尸”连接。

消费者管理TCP连接

何时创建TCP连接

和生产者不同的是，构建KafkaConsumer实例时是不会创建任何TCP连接的。TCP连接是在调用KafkaConsumer.poll方法时被创建的。再细粒度地说，在poll方法内部有3个时机可以创建TCP连接：

1.发起FindCoordinator请求时。

消费者端有个组件叫协调者（Coordinator），它驻留在Broker端的内存中，负责消费者组的组成员管理和各个消费者的位移提交管理。当消费者程序首次启动调用poll方法时，它需要向Kafka集群发送一个名为FindCoordinator的请求，向集群中当前负载最小的那台Broker发送请求。希望Kafka集群告诉它哪个Broker是管理它的协调者。

2.连接协调者时。

Broker处理完上一步发送的FindCoordinator请求之后，会返还对应的响应结果（Response），显式地告诉消费者哪个Broker是真正的协调者，因此在这一步，消费者知晓了真正的协调者后，会创建连向该 Broker的Socket连接。只有成功连入协调者，协调者才能开启正常的组协调操作，比如加入组、等待组分配方案、心跳请求处理、位移获取、位移提交等。

3.消费数据时。

消费者会为每个要消费的分区创建与该分区领导者副本所在Broker连接的TCP。举个例子，假设消费者要消费5个分区的数据，这5个分区各自的领导者副本分布在4台Broker上，那么该消费者在消费时会创建与这4台Broker的Socket连接。

注意：当第三类TCP连接成功创建后，消费者程序就会废弃第一类TCP连接，第一类TCP连接会在后台被默默地关闭掉。对一个运行了一段时间的消费者程序来说，只会有后面两类TCP连接存在。

通过日志查看：

[2019-05-27 10:00:54,142] DEBUG [ConsumerclientId=consumer-1, groupId=test] Initiating connection to nodelocalhost:9092 (id: -1 rack: null) using address localhost/127.0.0.1
(org.apache.kafka.clients.NetworkClient:944)
…
[2019-05-27 10:00:54,188] DEBUG [ConsumerclientId=consumer-1, groupId=test] Sending metadata request MetadataRequestData(topics=[MetadataRequestTopic(name=‘t4’)],
allowAutoTopicCreation=true, includeClusterAuthorizedOperations=false, includeTopicAuthorizedOperations=false) to nodelocalhost:9092 (id: -1 rack: null) (org.apache.kafka.clients.NetworkClient:1097)
…
[2019-05-27 10:00:54,188] TRACE [ConsumerclientId=consumer-1, groupId=test] Sending FIND_COORDINATOR {key=test,key_type=0} with correlation id 0 to node-1
(org.apache.kafka.clients.NetworkClient:496)
[2019-05-27 10:00:54,203] TRACE [ConsumerclientId=consumer-1, groupId=test] Completed receivefrom node-1 for FIND_COORDINATORwith correlation id 0, received
{throttle_time_ms=0,error_code=0,error_message=null, node_id=2,host=localhost,port=9094}(org.apache.kafka.clients.NetworkClient:837)
…
[2019-05-27 10:00:54,204] DEBUG [ConsumerclientId=consumer-1, groupId=test] Initiating connection to nodelocalhost:9094 (id: 2147483645 rack: null) using address localhost/127.0.0.1
(org.apache.kafka.clients.NetworkClient:944)
…
[2019-05-27 10:00:54,237] DEBUG [ConsumerclientId=consumer-1, groupId=test] Initiating connection to nodelocalhost:9094 (id: 2 rack: null) using address localhost/127.0.0.1
(org.apache.kafka.clients.NetworkClient:944)
[2019-05-27 10:00:54,237] DEBUG [ConsumerclientId=consumer-1, groupId=test] Initiating connection to nodelocalhost:9092 (id: 0 rack: null) using address localhost/127.0.0.1
(org.apache.kafka.clients.NetworkClient:944)
[2019-05-27 10:00:54,238] DEBUG [ConsumerclientId=consumer-1, groupId=test] Initiating connection to nodelocalhost:9093 (id: 1 rack: null) using address localhost/127.0.0.1
(org.apache.kafka.clients.NetworkClient:944)

日志的第一行是消费者程序创建的第一个TCP连接，就像我们前面说的，这个Socket用于发送FindCoordinator请求。由于这是消费者程序创建的第一个连接，此时消费者对于要连接的Kafka集群一无所知，因此它连接的Broker节点的ID是-1，表示消费者尚未获取到Broker数据。

日志的第二行，消费者复用了刚才创建的那个Socket连接，向Kafka集群发送元数据请求以获取整个集群的信息。

日志的第三行表明，消费者程序开始发送FindCoordinator请求给第一步中连接的Broker，即localhost:9092，也就是nodeId等于-1的那个。在十几毫秒之后，消费者程序成功地获悉协调者所在的Broker信息，也就是第四行的“node_id = 2”。

完成这些之后，消费者就已经知道协调者Broker的连接信息了，因此在日志的第五行发起了第二个Socket连接，创建了连向localhost:9094的TCP。只有连接了协调者，消费者进程才能正常地开启消费者组的各种功能以及后续的消息消费。

在日志的最后三行中，消费者又分别创建了新的TCP连接，主要用于实际的消息获取。要消费的分区的领导者副本在哪台Broker上，消费者就要创建连向哪台Broker的TCP。

那么2147483645是怎么来的呢？

它是由Integer.MAX_VALUE减去协调者所在Broker的真实ID计算得来的。看第四行的内容，我们可以知道协调者ID是2，因此这个Socket连接的节点ID就是 Integer.MAX_VALUE减去2，即2147483647减去2，也就是2147483645。这种节点ID的标记方式目的就是要让组协调请求和真正的数据获取请求使用不同的Socket连接。至于后面的0、1、2，那就很好解释了。它们表征了真实的Broker ID，也就是我们在server.properties中配置的broker.id值。