21 | Java 消费者是如何管理TCP连接的?

胡夕



该思维导图由 AI 生成，仅供参考

你好，我是胡夕。今天我要和你分享的主题是：Kafka 的 Java 消费者是如何管理 TCP 连接的。
在专栏第 13 讲中，我们专门聊过“Java生产者是如何管理 TCP 连接资源的”这个话题，你应该还有印象吧？今天算是它的姊妹篇，我们一起来研究下 Kafka 的 Java消费者管理 TCP 或 Socket 资源的机制。只有完成了今天的讨论，我们才算是对 Kafka 客户端的 TCP 连接管理机制有了全面的了解。
和之前一样，我今天会无差别地混用 TCP 和 Socket 两个术语。毕竟，在 Kafka 的世界中，无论是 ServerSocket，还是 SocketChannel，它们实现的都是 TCP 协议。或者这么说，Kafka 的网络传输是基于 TCP 协议的，而不是基于 UDP 协议，因此，当我今天说到 TCP 连接或 Socket 资源时，我指的是同一个东西。
何时创建 TCP 连接？我们先从消费者创建 TCP 连接开始讨论。消费者端主要的程序入口是 KafkaConsumer 类。和生产者不同的是，构建 KafkaConsumer 实例时是不会创建任何 TCP 连接的，也就是说，当你执行完 new KafkaConsumer(properties) 语句后，你会发现，没有 Socket 连接被创建出来。这一点和 Java 生产者是有区别的，主要原因就是生产者入口类 KafkaProducer 在构建实例的时候，会在后台默默地启动一个 Sender 线程，这个 Sender 线程负责 Socket 连接的创建。

公开

同步至部落

取消

完成

0/2000

荧光笔

直线

曲线

笔记

复制

AI

深入了解
翻译
英语
中文简体
中文繁体
法语
德语
日语
韩语
俄语
西班牙语
阿拉伯语
解释
总结

Kafka Java消费者TCP连接管理机制深入探讨了Kafka的Java消费者如何管理TCP连接。文章首先介绍了TCP连接的创建时机，包括发起FindCoordinator请求、连接协调者以及消费数据时。接着讨论了TCP连接的数量，并通过具体的Kafka日志验证了消费者创建的TCP连接数量和节点ID的变化。总结指出，消费者程序通常会创建三类TCP连接，分别用于确定协调者和获取集群元数据、连接协调者执行组成员管理操作以及执行实际的消息获取。此外，文章还提出了一个问题：这些TCP连接的生命周期是否相同，以及它们何时被关闭。另外，文章还探讨了消费者关闭TCP连接的方式以及可能的问题，并提出了对Kafka的设计方案的思考。最后，开放了一个讨论问题，引发读者思考。整体而言，本文详实地探讨了Kafka Java消费者管理TCP连接的机制，对于想要深入了解该方面知识的读者来说，是一篇值得阅读的文章。

仅可试看部分内容，如需阅读全部内容，请付费购买文章所属专栏
《Kafka 核心技术与实战》，新⼈⾸单¥68

立即购买

登录后留言

全部留言(37)

最新
精选

AAA_叶子
消费者tcp连接一旦断开，就会导致rebalance，实际开发过程中，是不是需要尽量保证长连接的模式？
作者回复: 嗯，如果就是要长时间的消费，维持一个长连接是不错的选择
2019-12-18
3
22
taj3991
老师同一个消费组的客户端都只会连接到一个协调者吗？
作者回复: 是的。每个group都有一个与之对应的coordinator
2019-07-22

20
Geek_ab3d9a
老师您好，请问k8s这样的容器平台，适合部署kafka的消费者吗？如果容器平台起了二个一模一样的消费者，对kafka来说会不会不知道自己通信的哪一个消费者? kafka通过什么来判断不同的客户端?
作者回复: 每个客户端至少主机名和端口是不一样的，我是指在TCP连接这个层面。另外对于producer而言，其实Kafka也不用区分它们，反正知道它们都再向集群发送消息就行了。对于consumer而言，主要还是看group.id的设置以确定它们是否在同一个group。最后如果是你想区分客户端，那么可以设置不同的client.id
2020-03-17

8
天天向上
元数据不包含协调者信息吗？为啥还要再请求一次协调者信息什么设计思路？
作者回复: 不包括，因为你请求元数据的broker可能不是Coordinator，没有Coordinator的信息
2019-09-19

7
小木匠
“负载是如何评估的呢？其实很简单，就是看消费者连接的所有 Broker 中，谁的待发送请求最少。” 老师这个没太明白，这时候消费者不是还没连接么？那这部分信息是从哪获取到的呢？消费者本地吗？
作者回复: 刚开始的时候当然就类似于随机选broker了，但后面慢慢积累了一些数据之后这个小优化还是会起一些作用的
2019-07-22

5
真锅
意思是即便知道了协调者在node 2上，还是会依然用2147483645这个id的TCP连接去跟协调者通信吗。
作者回复: 这个数字不是固定的，而是用MAX - broker ID算出来的。对Coordinator的连接来说，是的！它的ID就是这个算法
2020-04-25

4
Eco
应该是3个tcp连接，第一个id=-1的没什么争议，然后是连接协调者的，但是broker，5个分区的leader肯定会分布到这两台broker上，那么第三类tcp就是2个tcp连接，但是这2个中完全可以有一个是直接使用连接协调者的那个tcp连接吧，但老师好像说过连接协调者的连接会和传输数据的分开，id的计算都不相同，好吧，那就4个tcp连接吧。可这里真的不能复用吗？我觉得可以。
作者回复: 目前与Coordinator和普通数据交互的TCP连接的确是分开的，你要说是否能复用，我觉得当然可以复用，只不过现在没有这么设计：）
2020-01-19
2
4
yes
老师我有个疑问，consumer在FindCoordinator的时候会选择负载最小的broker进行连接，文章说看消费者连接的所有 Broker 中，谁的待发送请求最少。请问consumer如何得知这个消息？如果它想知道这个消息，不就得先和“某个东西”建立连接了？
作者回复: 它首先倾向于使用已建立连接的节点。如果已建立连接的节点都在使用中，可能会创建新的TCP连接
2020-06-20
2
3
Treagzhao
老师，“消费者程序会向集群中当前负载最小的那台 Broker 发送请求”，消费者怎么单方面知道服务器待发送的消息数量呢？而且应该只有leader才会实际发送消息吧，follower待发送的都是0,消费者怎么在建立连接之前就知道服务器的角色呢？
作者回复: 是这样判断的，就是看消费者与broker的TCP连接上的待处理请求的个数
2020-05-19

2
臧萌
我们要设计一个消息系统。有两个选择，更好的一种是每种不同schema的消息发一个topic。但是有一种担心是consumer会为每个topic建立一个连接，造成连接数太多。请问胡老师，kafka client的consumer是每个集群固定数目的tcp连接，还是和topic数目相关？
作者回复: 和它要订阅的topic分区数以及这些分区在broker上的散列情况有关。比如你订阅了100个分区，但这个100个分区的leader副本都在一个broker上，那么长期来看consumer也就只和这1个broker建立连接；相反如果这100分区散列在100个broker上，那么长期来看consumer会和100个broker维持长连接
2019-08-29
2
2

收起评论