如何在网站投放广告,遵义本地网络平台,WordPress 弹出二维码,哪种语言做的网站好目录
前言#xff1a;
Kafka生产者程序 Kafka生产者客户端如何创建TCP连接 Kafka生产者客户端如何关闭TCP连接
总结#xff1a;
参考资料 前言#xff1a; 在网络层协议中#xff0c;TCP作用在第四层传输层、Http协议作用在第七层最上层应用层#xff0c;一个完整的…目录
前言
Kafka生产者程序 Kafka生产者客户端如何创建TCP连接 Kafka生产者客户端如何关闭TCP连接
总结
参考资料 前言 在网络层协议中TCP作用在第四层传输层、Http协议作用在第七层最上层应用层一个完整的网络传输信息会优先到达第四层然后在往上传输到第七层TCP协议相比于Http协议提供更好的连接稳定性及TCP提供的多路复用请求及可靠的消息交付语义保证如自动重传丢失的报文等kafka在设计上使用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 (ProducerString, String producer new KafkaProducer(props)) {producer.send(new ProducerRecordString, String(……), callback);……
} 当我们开发一个Producer应用时生产者会向Kafka集群中指定的主题Topic发送消息这必然涉及与Kafka Broker创建TCP连接。那么Kafka的Producer客户端是如何管理这些TCP连接的呢 Kafka生产者客户端如何创建TCP连接 要回答上面这个问题我们首先要弄明白生产者代码是什么时候创建TCP连接的。就上面的那段代码而言可能创建TCP连接的地方有两处Producer producer new KafkaProducer(props)和producer.send(msg, callback)。你觉得连向Broker端的TCP连接会是哪里创建的呢前者还是后者抑或是两者都有请先思考5秒钟然后我给出我的答案。 首先生产者应用在创建KafkaProducer实例时是会建立与Broker的TCP连接的。其实这种表述也不是很准确应该这样说在创建KafkaProducer实例时生产者应用会在后台创建并启动一个名为Sender的线程该Sender线程开始运行时首先会创建与Broker的连接。我截取了一段测试环境中的日志来说明这一点
[2018-12-09 09:35:45,620] DEBUG [Producer clientIdproducer-1] Initialize connection to node localhost:9093 (id: -2 rack: null) for sending metadata request (org.apache.kafka.clients.NetworkClient:1084)[2018-12-09 09:35:45,622] DEBUG [Producer clientIdproducer-1] Initiating connection to node localhost:9093 (id: -2 rack: null) using address localhost/127.0.0.1 (org.apache.kafka.clients.NetworkClient:914)[2018-12-09 09:35:45,814] DEBUG [Producer clientIdproducer-1] Initialize connection to node localhost:9092 (id: -1 rack: null) for sending metadata request (org.apache.kafka.clients.NetworkClient:1084)[2018-12-09 09:35:45,815] DEBUG [Producer clientIdproducer-1] Initiating connection to node localhost:9092 (id: -1 rack: null) using address localhost/127.0.0.1 (org.apache.kafka.clients.NetworkClient:914)[2018-12-09 09:35:45,828] DEBUG [Producer clientIdproducer-1] Sending metadata request (typeMetadataRequest, topics) to node localhost:9093 (id: -2 rack: null) (org.apache.kafka.clients.NetworkClient:1068) 你也许会问怎么可能是这样如果不调用send方法这个Producer都不知道给哪个主题发消息它又怎么能知道连接哪个Broker呢难不成它会连接bootstrap.servers参数指定的所有Broker吗嗯是的Java Producer目前还真是这样设计的。 我在这里稍微解释一下bootstrap.servers参数。它是Producer的核心参数之一指定了这个Producer启动时要连接的Broker地址。请注意这里的“启动时”代表的是Producer启动时会发起与这些Broker的连接。因此如果你为这个参数指定了1000个Broker连接信息那么很遗憾你的Producer启动时会首先创建与这1000个Broker的TCP连接。 在实际使用过程中我并不建议把集群中所有的Broker信息都配置到bootstrap.servers中通常你指定34台就足以了。因为Producer一旦连接到集群中的任一台Broker就能拿到整个集群的Broker信息故没必要为bootstrap.servers指定所有的Broker。 从上面这段日志中我们可以发现在KafkaProducer实例被创建后以及消息被发送前Producer应用就开始创建与两台Broker的TCP连接了。当然了在我的测试环境中我为bootstrap.servers配置了localhost:9092、localhost:9093来模拟不同的Broker但是这并不影响后面的讨论。另外日志输出中的最后一行也很关键它表明Producer向某一台Broker发送了METADATA请求尝试获取集群的元数据信息——这就是前面提到的Producer能够获取集群所有信息的方法。
纵然KafkaProducer是线程安全的我也不赞同创建KafkaProducer实例时启动Sender线程的做法。写了《Java并发编程实践》的那位布赖恩·格茨Brian Goetz大神明确指出了这样做的风险在对象构造器中启动线程会造成this指针的逃逸。理论上Sender线程完全能够观测到一个尚未构造完成的KafkaProducer实例。当然在构造对象时创建线程没有任何问题但最好是不要同时启动它。 针对TCP连接何时创建的问题目前我们的结论是这样的TCP连接是在创建KafkaProducer实例时建立的。那么我们想问的是它只会在这个时候被创建吗
当然不是TCP连接还可能在两个地方被创建一个是在更新元数据后另一个是在消息发送时。为什么说是可能因为这两个地方并非总是创建TCP连接。当Producer更新了集群的元数据信息之后如果发现与某些Broker当前没有连接那么它就会创建一个TCP连接。同样地当要发送消息时Producer发现尚不存在与目标Broker的连接也会创建一个。 接下来我们来看看Producer更新集群元数据信息的两个场景。
场景一当Producer尝试给一个不存在的主题发送消息时Broker会告诉Producer说这个主题不存在。此时Producer会发送METADATA请求给Kafka集群去尝试获取最新的元数据信息。
场景二Producer通过metadata.max.age.ms参数定期地去更新元数据信息。该参数的默认值是300000即5分钟也就是说不管集群那边是否有变化Producer每5分钟都会强制刷新一次元数据以保证它是最及时的数据。
讲到这里我们可以“挑战”一下社区对Producer的这种设计的合理性。目前来看一个Producer默认会向集群的所有Broker都创建TCP连接不管是否真的需要传输请求。这显然是没有必要的。再加上Kafka还支持强制将空闲的TCP连接资源关闭这就更显得多此一举了。
试想一下在一个有着1000台Broker的集群中你的Producer可能只会与其中的35台Broker长期通信但是Producer启动后依次创建与这1000台Broker的TCP连接。一段时间之后大约有995个TCP连接又被强制关闭。这难道不是一种资源浪费吗很显然这里是有改善和优化的空间的。 Kafka生产者客户端如何关闭TCP连接
Producer端关闭TCP连接的方式有两种一种是用户主动关闭一种是Kafka自动关闭。
我们先说第一种。这里的主动关闭实际上是广义的主动关闭甚至包括用户调用kill -9主动“杀掉”Producer应用。当然最推荐的方式还是调用producer.close()方法来关闭。
第二种是Kafka帮你关闭这与Producer端参数connections.max.idle.ms的值有关。默认情况下该参数值是9分钟即如果在9分钟内没有任何请求“流过”某个TCP连接那么Kafka会主动帮你把该TCP连接关闭。用户可以在Producer端设置connections.max.idle.ms-1禁掉这种机制。一旦被设置成-1TCP连接将成为永久长连接。当然这只是软件层面的“长连接”机制由于Kafka创建的这些Socket连接都开启了keepalive因此keepalive探活机制还是会遵守的。
值得注意的是在第二种方式中TCP连接是在Broker端被关闭的但其实这个TCP连接的发起方是客户端因此在TCP看来这属于被动关闭的场景即passive close。被动关闭的后果就是会产生大量的CLOSE_WAIT连接因此Producer端或Client端没有机会显式地观测到此连接已被中断。
总结
KafkaProducer实例创建时启动Sender线程从而创建与bootstrap.servers中所有Broker的TCP连接。KafkaProducer实例首次更新元数据信息之后还会再次创建与集群中所有Broker的TCP连接。如果Producer端发送消息到某台Broker时发现没有与该Broker的TCP连接那么也会立即创建连接。如果设置Producer端connections.max.idle.ms参数大于0则步骤1中创建的TCP连接会被自动关闭如果设置该参数-1那么步骤1中创建的TCP连接将无法被关闭从而成为“僵尸”连接。
参考资料
极客时间课程《Kafka核心技术与实战》
13.Java生产者是如何管理TCP连接的
