码迷,mamicode.com
首页 > Web开发 > 详细

kafka源码分析(二)Metadata的数据结构与读取、更新策略

时间:2018-05-22 20:47:33      阅读:297      评论:0      收藏:0      [点我收藏+]

标签:分享图片   信息   失败   fms   rem   线程安全   循环   后台   技术   

技术分享图片

 

一、基本思路

 异步发送的基本思路就是:send的时候,KafkaProducer把消息放到本地的消息队列RecordAccumulator,然后一个后台线程Sender不断循环,把消息发给Kafka集群。

要实现这个,还得有一个前提条件:就是KafkaProducer/Sender都需要获取集群的配置信息Metadata。所谓Metadata,也就是在上一篇所讲的,Topic/Partion与broker的映射关系:每一个Topic的每一个Partion,得知道其对应的broker列表是什么,其中leader是谁,follower是谁。

 

二、2个数据流

 

所以在上图中,有2个数据流: 

Metadata流(A1,A2,A3):Sender从集群获取信息,然后更新Metadata; KafkaProducer先读取Metadata,然后把消息放入队列。

 

消息流(B1, B2, B3)

从上图可以看出,Metadata是多个producer线程读,一个sender线程更新,因此它必须是线程安全的

 

三、Metadata的线程安全性

从下面代码也可以看出,它的所有public方法都是synchronized:

 1 public final class Metadata {
 2   。。。
 3     public synchronized Cluster fetch() {
 4         return this.cluster;
 5     }
 6     public synchronized long timeToNextUpdate(long nowMs) {
 7        。。。
 8     }
 9     public synchronized int requestUpdate() {
10       。。。
11     }
12     。。。    
13

 

 

四、Metadata的数据结构

 1 public final class Metadata {
 2 ...
 3     private final long refreshBackoffMs;  //更新失败的情况下,下1次更新的补偿时间(这个变量在代码中意义不是太大)
 4     private final long metadataExpireMs; //关键值:每隔多久,更新一次。缺省是600*1000,也就是10分种
 5     private int version;         //每更新成功1次,version递增1。这个变量主要用于在while循环,wait的时候,作为循环判断条件
 6     private long lastRefreshMs;  //上一次更新时间(也包含更新失败的情况)
 7     private long lastSuccessfulRefreshMs; //上一次成功更新的时间(如果每次都成功的话,则2者相等。否则,lastSuccessulRefreshMs < lastRefreshMs)
 8     private Cluster cluster;   //集群配置信息
 9     private boolean needUpdate;  //是否强制刷新
10 11   ...
12 }

 

kafka源码分析(二)Metadata的数据结构与读取、更新策略

标签:分享图片   信息   失败   fms   rem   线程安全   循环   后台   技术   

原文地址:https://www.cnblogs.com/zcjcsl/p/8746561.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!