// 用于添加Metrics的tags
val tags = Map("topic" -> topicAndPartition.topic, "partition" -> topicAndPartition.partition.toString)

// 注册Gauge度量对象
newGauge("NumLogSegments",
new Gauge[Int] {
    // 记录当前Log中的Segment对象个数
    def value = numberOfSegments
},
tags)
...
def numberOfSegments: Int = segments.size

...
// 使用Meter度量类统计全部分区的ISR集合发生扩张/缩减的频率
val isrExpandRate = newMeter("IsrExpandsPerSec",  "expands", TimeUnit.SECONDS)
val isrShrinkRate = newMeter("IsrShrinksPerSec",  "shrinks", TimeUnit.SECONDS)
...

// ISR集合扩充
def maybeExpandIsr(replicaId: Int) {
  ...
  // 这里会使用Meter度量对象标识一次ISR集合的扩张
  replicaManager.isrExpandRate.mark()
  ...
}
// ISR集合缩减
def maybeShrinkIsr(replicaMaxLagTimeMs: Long) {
  ...
  // 这里会使用Meter度量对象标识一次ISR集合的缩减
  replicaManager.isrShrinkRate.mark()
  ...
}

class RequestMetrics(name: String) extends KafkaMetricsGroup {
  val tags = Map("request" -> name)
  // 统计每秒请求数
  val requestRate = newMeter("RequestsPerSec", "requests", TimeUnit.SECONDS, tags)
  // time a request spent in a request queue
  // 统计Request在RequestChannel中的等待时间
  val requestQueueTimeHist = newHistogram("RequestQueueTimeMs", biased = true, tags)
  // time a request takes to be processed at the local broker
  // 统计Request在当前Broker中处理的用时
  val localTimeHist = newHistogram("LocalTimeMs", biased = true, tags)
  // time a request takes to wait on remote brokers (currently only relevant to fetch and produce requests)
  // 统计此Broker发送的Request在远端Broker中处理的用时，例如FetchRequest
  val remoteTimeHist = newHistogram("RemoteTimeMs", biased = true, tags)
  // time a request is throttled (only relevant to fetch and produce requests)
  val throttleTimeHist = newHistogram("ThrottleTimeMs", biased = true, tags)
  // time a response spent in a response queue
  // 统计Response在RequestChannel中的等待时间
  val responseQueueTimeHist = newHistogram("ResponseQueueTimeMs", biased = true, tags)
  // time to send the response to the requester
  // 统计发送Response的用时
  val responseSendTimeHist = newHistogram("ResponseSendTimeMs", biased = true, tags)
  // 统计总耗时
  val totalTimeHist = newHistogram("TotalTimeMs", biased = true, tags)
}

// Request请求Case类
case class Request(processor: Int, connectionId: String, session: Session, private var buffer: ByteBuffer, startTimeMs: Long, securityProtocol: SecurityProtocol) {
  // These need to be volatile because the readers are in the network thread and the writers are in the request
  // handler threads or the purgatory threads
  // @volatile修饰保证线程可见性；记录操作时间的几个字段
  @volatile var requestDequeueTimeMs = -1L
  @volatile var apiLocalCompleteTimeMs = -1L
  @volatile var responseCompleteTimeMs = -1L
  @volatile var responseDequeueTimeMs = -1L
  @volatile var apiRemoteCompleteTimeMs = -1L
    ...
}

class KafkaRequestHandler(id: Int,
                          brokerId: Int,
                          val aggregateIdleMeter: Meter,
                          val totalHandlerThreads: Int,
                          val requestChannel: RequestChannel,
                          apis: KafkaApis) extends Runnable with Logging {
  ...
  def run() {
    // 从RequestChannel的requestQueue队列中获取请求
    req = requestChannel.receiveRequest(300)
    ...
    // 更新requestDequeueTimeMs
    req.requestDequeueTimeMs = SystemTime.milliseconds
    ...
    // 使用KafkaApis处理RequestChannel.Request
    apis.handle(req)
    ...
  }
  ...
}

private def processCompletedSends() {
  // 遍历completedSends队列
  selector.completedSends.asScala.foreach { send =>
    ...
    // 调用Request的updateRequestMetrics()更新度量值
    resp.request.updateRequestMetrics()
    ...
  }
}

class KafkaServer(val config: KafkaConfig, time: Time = SystemTime, threadNamePrefix: Option[String] = None) extends Logging with KafkaMetricsGroup {
  ...
  def startup() {
    ...
    /** register broker metrics
      * 创建并注册Broker的Timer Metrics对象
      * */
    registerStats()
    ...
  }
  ...
  private def registerStats() {
    BrokerTopicStats.getBrokerAllTopicsStats()
    ControllerStats.uncleanLeaderElectionRate
    ControllerStats.leaderElectionTimer
  }
  ...
}

object ControllerStats extends KafkaMetricsGroup {

  // 注册Meter度量对象
  private val _uncleanLeaderElectionRate = newMeter("UncleanLeaderElectionsPerSec", "elections", TimeUnit.SECONDS)
  // 注册Timer度量对象
  private val _leaderElectionTimer = new KafkaTimer(newTimer("LeaderElectionRateAndTimeMs", TimeUnit.MILLISECONDS, TimeUnit.SECONDS))

  // KafkaServer needs to initialize controller metrics during startup. We perform initialization
  // through method calls to avoid Scala compiler warnings.
  def uncleanLeaderElectionRate: Meter = _uncleanLeaderElectionRate

  def leaderElectionTimer: KafkaTimer = _leaderElectionTimer
}

class BrokerChangeListener() extends IZkChildListener with Logging {
  this.logIdent = "[BrokerChangeListener on Controller " + controller.config.brokerId + "]: "
  def handleChildChange(parentPath : String, currentBrokerList : java.util.List[String]) {
    inLock(controllerContext.controllerLock) {
      if (hasStarted.get) { // 需要在状态机启动的情况下执行
        ControllerStats.leaderElectionTimer.time { // 计时操作，用于JMX监控
        ...
      }
      ...
    }
    ...
  }
  ...
}

/**
 * A SampledStat records a single scalar value measured over one or more samples. Each sample is recorded over a
 * configurable window. The window can be defined by number of events or elapsed time (or both, if both are given the
 * window is complete when <i>either</i> the event count or elapsed time criterion is met).
 * <p>
 * All the samples are combined to produce the measurement. When a window is complete the oldest sample is cleared and
 * recycled to begin recording the next sample.
 * 
 * Subclasses of this class define different statistics measured using this basic pattern.
 *
 * 表示一个抽样的度量值，除了Total和Rate外的其他MeasurableStat接口实现都依赖它的功能
 */
public abstract class SampledStat implements MeasurableStat {
    // 指定每个样本的初始值
    private double initialValue;
    // 当前使用的Sample的下标
    private int current = 0;
    // List类型，保存当前SampledStat中的多个Sample
    protected List<Sample> samples;
    // 构造方法
    public SampledStat(double initialValue) {
        this.initialValue = initialValue;
        this.samples = new ArrayList<Sample>(2);
    }
    ...
}

protected static class Sample {
    // 指定样本的初始值
    public double initialValue;
    // 记录当前样本的事件数
    public long eventCount;
    // 记录当前样本的时间窗口开始的时间戳
    public long lastWindowMs;
    // 记录样本的值
    public double value;

    public Sample(double initialValue, long now) {
        this.initialValue = initialValue;
        this.eventCount = 0;
        this.lastWindowMs = now;
        this.value = initialValue;
    }

    public void reset(long now) {
        this.eventCount = 0;
        this.lastWindowMs = now;
        this.value = initialValue;
    }

    // 检测eventCount和lastWindows决定当前样本是否已经取样完成
    public boolean isComplete(long timeMs, MetricConfig config) {
        return timeMs - lastWindowMs >= config.timeWindowMs() || // 检测时间窗口
                eventCount >= config.eventWindow(); // 检测事件数
    }
}

/**
 * Configuration values for metrics
 */
public class MetricConfig {
    // 定义了度量值的上下限，超过范围后会抛出异常
    private Quota quota;
    // 指定样本的个数，默认值是2
    private int samples;
    // 指定每个样本中事件的上限，当一个样本中的事件数超过此值后，则开始下个样本的记录，默认值是Long.MAX_VALUE
    private long eventWindow;
    // 指定每个样本取样的时间窗口，当一个样本的时间窗口超出范围后，则开始下个样本的记录，默认值是30秒
    private long timeWindowMs;
    // 指定相关的Tag标签
    private Map<String, String> tags;
    ...
}

// 会将过期的Sample重置，调用combine()方法完成计算，该方法是抽象方法
@Override
public double measure(MetricConfig config, long now) {
    purgeObsoleteSamples(config, now);
    return combine(this.samples, config, now);
}

/* Timeout any windows that have expired in the absence of any events */
protected void purgeObsoleteSamples(MetricConfig config, long now) {
    // 计算过期时长
    long expireAge = config.samples() * config.timeWindowMs();
    for (int i = 0; i < samples.size(); i++) {
        Sample sample = this.samples.get(i);
        if (now - sample.lastWindowMs >= expireAge)
            // 检测到Sample过期，将其重置
            sample.reset(now);
    }
}

/**
 * An un-windowed cumulative total maintained over all time.
 * Total统计的值单调递增
 */
public class Total implements MeasurableStat {

    private double total;

    public Total() {
        this.total = 0.0;
    }

    public Total(double value) {
        this.total = value;
    }

    // 在该方法中完成累加
    @Override
    public void record(MetricConfig config, double value, long now) {
        this.total += value;
    }

    // 在该方法中返回累加值
    @Override
    public double measure(MetricConfig config, long now) {
        return this.total;
    }

}

/**
 * A {@link SampledStat} that maintains a simple count of what it has seen.
 * Count与Total的不同之处在于，随着时间窗口的推移，Count中会有Sample过期，所以Count记录的是一段时间内的总数，而Total的值则单调递增。
 */
public class Count extends SampledStat {

    public Count() {
        super(0);
    }

    // 在该方法中完成Sample的value累加
    @Override
    protected void update(Sample sample, MetricConfig config, double value, long now) {
        sample.value += 1.0;
    }

    // 在该方法中将所有未过期的Sample的value进行求和并返回
    @Override
    public double combine(List<Sample> samples, MetricConfig config, long now) {
        double total = 0.0;
        for (int i = 0; i < samples.size(); i++)
            total += samples.get(i).value;
        return total;
    }

}

/**
 * A {@link SampledStat} that gives the max over its samples.
 *
 * 记录最大值
 */
public final class Max extends SampledStat {

    public Max() {
        super(Double.NEGATIVE_INFINITY);
    }

    // 每个比较新的值和旧的值的大小，记录最大的那个
    @Override
    protected void update(Sample sample, MetricConfig config, double value, long now) {
        sample.value = Math.max(sample.value, value);
    }

    @Override
    public double combine(List<Sample> samples, MetricConfig config, long now) {
        double max = Double.NEGATIVE_INFINITY;
        // 得到所有Sample中的最大值
        for (int i = 0; i < samples.size(); i++)
            max = Math.max(max, samples.get(i).value);
        return max;
    }

}

/**
 * A {@link SampledStat} that gives the min over its samples.
 */
public class Min extends SampledStat {

    public Min() {
        super(Double.MIN_VALUE);
    }

    // 每个比较新的值和旧的值的大小，记录最小的那个
    @Override
    protected void update(Sample sample, MetricConfig config, double value, long now) {
        sample.value = Math.min(sample.value, value);
    }

    @Override
    public double combine(List<Sample> samples, MetricConfig config, long now) {
        double max = Double.MAX_VALUE;
        // 得到所有Sample中的最小值
        for (int i = 0; i < samples.size(); i++)
            max = Math.min(max, samples.get(i).value);
        return max;
    }

}

/**
 * A {@link SampledStat} that maintains a simple average over its samples.
 */
public class Avg extends SampledStat {

    public Avg() {
        super(0.0);
    }

    // 对每次新的值进行累加
    @Override
    protected void update(Sample sample, MetricConfig config, double value, long now) {
        sample.value += value;
    }

    @Override
    public double combine(List<Sample> samples, MetricConfig config, long now) {
        double total = 0.0;
        long count = 0;
        // 将所有Sample的值进行累加，并将所有Sample的事件数进行累加
        for (int i = 0; i < samples.size(); i++) {
            Sample s = samples.get(i);
            total += s.value;
            count += s.eventCount;
        }

        // 计算平均值
        return count == 0 ? 0 : total / count;
    }

}

public final class MetricName {
    // KafkaMetric的名称
    private final String name;
    // 记录所属的逻辑组的名称
    private final String group;
    // 对KafkaMetric的描述信息
    private final String description;
    // 记录额外的键值对信息
    private Map<String, String> tags;
    ...
}

public final class Sensor {

    private final Metrics registry;
    // 当前Sensor对象的名称，Metrics通过该名称区别不同的Sensor对象
    private final String name;
    // Sensor是可以分为多层的，该字段指定了当前Sensor对象的父Sensor
    private final Sensor[] parents;
    // 保存构成Sensor的度量对象
    private final List<Stat> stats;
    /**
     * 保存构成Sensor的KafkaMetric对象，
     * 当度量对象添加进Sensor时会创建对应的KafkaMetric对象（CompoundStat会创建多个），
     * 并保存到此集合中
     */
    private final List<KafkaMetric> metrics;
    // 默认的配置信息
    private final MetricConfig config;
    private final Time time;
    // 最后一次执行record()方法的时间戳
    private volatile long lastRecordTime;
    /**
     * 长时间未使用Sensor会被认为是过期Sensor，
     * 由ExpireSensorTask线程负责进行清理，此字段记录成为过期Sensor”的阈值
     */
    private final long inactiveSensorExpirationTimeMs;
    ...
}

public synchronized void add(MetricName metricName, MeasurableStat stat, MetricConfig config) {
    // 创建KafkaMetric对象
    KafkaMetric metric = new KafkaMetric(new Object(),
                                         Utils.notNull(metricName),
                                         Utils.notNull(stat),
                                         config == null ? this.config : config,
                                         time);
    // 将KafkaMetric保存到Metrics中，创建并注册对应的MBean
    this.registry.registerMetric(metric);
    // 添加到metrics集合
    this.metrics.add(metric);
    // 添加到stats集合
    this.stats.add(stat);
}

public synchronized void add(CompoundStat stat, MetricConfig config) {
    this.stats.add(Utils.notNull(stat));
    // 遍历CompoundStat中每个子Stat对象
    for (NamedMeasurable m : stat.stats()) {
        // 创建KafkaMetric对象
        KafkaMetric metric = new KafkaMetric(this, m.name(), m.stat(), config == null ? this.config : config, time);
        // 将KafkaMetric保存到Metrics中，创建并注册对应的MBean
        this.registry.registerMetric(metric);
        // 添加到metrics集合
        this.metrics.add(metric);
    }
}

/**
 * Record a value at a known time. This method is slightly faster than {@link #record(double)} since it will reuse
 * the time stamp.
 * @param value The value we are recording
 * @param timeMs The current POSIX time in milliseconds
 * @throws QuotaViolationException if recording this value moves a metric beyond its configured maximum or minimum
 *         bound
 */
public void record(double value, long timeMs) {
    this.lastRecordTime = timeMs;
    synchronized (this) {
        // increment all the stats
        // 调用stats集合中每个stat对象的record()方法
        for (int i = 0; i < this.stats.size(); i++)
            this.stats.get(i).record(config, value, timeMs);
        // 检测是否超出了MetricConfig指定的上下西限
        checkQuotas(timeMs);
    }
    for (int i = 0; i < parents.length; i++)
        // 调用每个父Sensor的record()方法
        parents[i].record(value, timeMs);
}

/**
 * Check if we have violated our quota for any metric that has a configured quota
 * @param timeMs
 */
private void checkQuotas(long timeMs) {
    for (int i = 0; i < this.metrics.size(); i++) {
        KafkaMetric metric = this.metrics.get(i);
        // 获取MetricConfig对象
        MetricConfig config = metric.config();
        if (config != null) {
            // 获取MetricsConfig对象中的Quota对象
            Quota quota = config.quota();
            if (quota != null) {
                // 计算最终的度量值
                double value = metric.value(timeMs);
                // 检测度量值是否超出上下限
                if (!quota.acceptable(value)) {
                    throw new QuotaViolationException(String.format(
                        "'%s' violated quota. Actual: %f, Threshold: %f",
                        metric.metricName(),
                        value,
                        quota.bound()));
                }
            }
        }
    }
}

public class Metrics implements Closeable {

    // 默认的配置信息
    private final MetricConfig config;
    // 保存了添加到Metrics中的KafkaMetric对象
    private final ConcurrentMap<MetricName, KafkaMetric> metrics;
    // 保存了添加到Metrics中的Sensor对象
    private final ConcurrentMap<String, Sensor> sensors;
    // 记录了每个Sensor的子Sensor集合
    private final ConcurrentMap<Sensor, List<Sensor>> childrenSensors;
    // 保存了使用的MetricsReporter对象，默认是JmxReporter
    private final List<MetricsReporter> reporters;
    private final Time time;
    // 用于执行ExpireSensorTask定时任务的线程池
    private final ScheduledThreadPoolExecutor metricsScheduler;
    ...
}

public synchronized Sensor sensor(String name, MetricConfig config, long inactiveSensorExpirationTimeSeconds, Sensor... parents) {
    // 根据name从sensors集合中获取Sensor对象
    Sensor s = getSensor(name);
    if (s == null) {
        // 创建Sensor对象
        s = new Sensor(this, name, parents, config == null ? this.config : config, time, inactiveSensorExpirationTimeSeconds);
        // 将新创建的Sensor对象保存到sensors集合中
        this.sensors.put(name, s);
        if (parents != null) {
            // 通过childrenSensors记录Sensor的层次关系
            for (Sensor parent : parents) {
                List<Sensor> children = childrenSensors.get(parent);
                if (children == null) {
                    children = new ArrayList<>();
                    childrenSensors.put(parent, children);
                }
                children.add(s);
            }
        }
        log.debug("Added sensor with name {}", name);
    }
    return s;
}

public synchronized void addReporter(MetricsReporter reporter) {
    Utils.notNull(reporter).init(new ArrayList<>(metrics.values()));
    this.reporters.add(reporter);
}

public synchronized void addMetric(MetricName metricName, MetricConfig config, Measurable measurable) {
    KafkaMetric m = new KafkaMetric(new Object(),
                                    Utils.notNull(metricName),
                                    Utils.notNull(measurable),
                                    config == null ? this.config : config,
                                    time);
    registerMetric(m);
}

synchronized void registerMetric(KafkaMetric metric) {
    MetricName metricName = metric.metricName();
    if (this.metrics.containsKey(metricName))
        throw new IllegalArgumentException("A metric named '" + metricName + "' already exists, can't register another one.");
    // 向metrics中添加KafkaMetric对象
    this.metrics.put(metricName, metric);
    for (MetricsReporter reporter : reporters)
        // 向每个MetricsReporter注册KafkaMetric对象
        reporter.metricChange(metric);
}

public Metrics(MetricConfig defaultConfig, List<MetricsReporter> reporters, Time time, boolean enableExpiration) {
    this.config = defaultConfig;
    this.sensors = new ConcurrentHashMap<>();
    this.metrics = new ConcurrentHashMap<>();
    this.childrenSensors = new ConcurrentHashMap<>();
    this.reporters = Utils.notNull(reporters);
    this.time = time;
    for (MetricsReporter reporter : reporters)
        reporter.init(new ArrayList<KafkaMetric>());

    // Create the ThreadPoolExecutor only if expiration of Sensors is enabled.
    // 创建ScheduledThreadPoolExecutor线程池
    if (enableExpiration) {
        this.metricsScheduler = new ScheduledThreadPoolExecutor(1);
        // Creating a daemon thread to not block shutdown
        this.metricsScheduler.setThreadFactory(new ThreadFactory() {
            public Thread newThread(Runnable runnable) {
                return Utils.newThread("SensorExpiryThread", runnable, true);
            }
        });
        // 启动定时任务
        this.metricsScheduler.scheduleAtFixedRate(new ExpireSensorTask(), 30, 30, TimeUnit.SECONDS);
    } else {
        this.metricsScheduler = null;
    }

    addMetric(metricName("count", "kafka-metrics-count", "total number of registered metrics"),
        // 创建一个记录metrics集合大小的Measurable对象并注册到Metrics中
        new Measurable() {
            @Override
            public double measure(MetricConfig config, long now) {
                return metrics.size();
            }
        });
}

class ExpireSensorTask implements Runnable {
    public void run() {
        // 遍历sensors集合
        for (Map.Entry<String, Sensor> sensorEntry : sensors.entrySet()) {
            // removeSensor also locks the sensor object. This is fine because synchronized is reentrant
            // There is however a minor race condition here. Assume we have a parent sensor P and child sensor C.
            // Calling record on C would cause a record on P as well.
            // So expiration time for P == expiration time for C. If the record on P happens via C just after P is removed,
            // that will cause C to also get removed.
            // Since the expiration time is typically high it is not expected to be a significant concern
            // and thus not necessary to optimize
            synchronized (sensorEntry.getValue()) {
                if (sensorEntry.getValue().hasExpired()) { // 检查Sensor是否过期
                    log.debug("Removing expired sensor {}", sensorEntry.getKey());
                    // 移除过期的Sensor对象
                    removeSensor(sensorEntry.getKey());
                }
            }
        }
    }
}

public interface MetricsReporter extends Configurable {

    /**
     * This is called when the reporter is first registered to initially register all existing metrics
     * 初始化函数，将MetricsReporter添加到Metrics时被调用，完成初始化操作
     * @param metrics All currently existing metrics
     */
    public void init(List<KafkaMetric> metrics);

    /**
     * This is called whenever a metric is updated or added
     * 向Metrics中添加KafkaMetric时被调用
     * @param metric
     */
    public void metricChange(KafkaMetric metric);

    /**
     * This is called whenever a metric is removed
     * 从Metrics中删除KafkaMetric时被调用
     * @param metric
     */
    public void metricRemoval(KafkaMetric metric);

    /**
     * Called when the metrics repository is closed.
     * 关闭操作
     */
    public void close();

}

private static class KafkaMbean implements DynamicMBean {
    // MBean的名称
    private final ObjectName objectName;
    // 记录了添加的KafkaMetric对象，这些KafkaMetric对象会被当作KafkaMbean对象的属性处理
    private final Map<String, KafkaMetric> metrics;

    public KafkaMbean(String mbeanName) throws MalformedObjectNameException {
        // 初始化metrics和objectName
        this.metrics = new HashMap<String, KafkaMetric>();
        this.objectName = new ObjectName(mbeanName);
    }

    public ObjectName name() {
        return objectName;
    }

    public void setAttribute(String name, KafkaMetric metric) {
        // 记录KafkaMetric对象
        this.metrics.put(name, metric);
    }

    @Override
    public Object getAttribute(String name) throws AttributeNotFoundException, MBeanException, ReflectionException {
        if (this.metrics.containsKey(name))
            // 返回指定KafkaMetric中的度量值
            return this.metrics.get(name).value();
        else
            throw new AttributeNotFoundException("Could not find attribute " + name);
    }

    @Override
    public AttributeList getAttributes(String[] names) {
        try {
            AttributeList list = new AttributeList();
            // 循环遍历names并调用getAttribute()方法
            for (String name : names)
                // 获取指定KafkaMetric中的度量值包装为Attribute对象并放入AttributeList
                list.add(new Attribute(name, getAttribute(name)));
            return list;
        } catch (Exception e) {
            log.error("Error getting JMX attribute: ", e);
            return new AttributeList();
        }
    }
    ...
}

// 将KafkaMetric以属性的形式添加到KafkaMBean中，并重新注册KafkaMBean
@Override
public void metricChange(KafkaMetric metric) {
    synchronized (LOCK) {
        // 添加KafkaMetric
        KafkaMbean mbean = addAttribute(metric);
        // 重新注册KafkaMBean
        reregister(mbean);
    }
}

private KafkaMbean addAttribute(KafkaMetric metric) {
    try {
        MetricName metricName = metric.metricName();
        // 获取KafkaMBean的名称
        String mBeanName = getMBeanName(metricName);
        // 如果没有该名称的KafkaMBean则创建
        if (!this.mbeans.containsKey(mBeanName))
            mbeans.put(mBeanName, new KafkaMbean(mBeanName));
        KafkaMbean mbean = this.mbeans.get(mBeanName);
        // 以属性的形式添加KafkaMetric
        mbean.setAttribute(metricName.name(), metric);
        return mbean;
    } catch (JMException e) {
        throw new KafkaException("Error creating mbean attribute for metricName :" + metric.metricName(), e);
    }
}

private void reregister(KafkaMbean mbean) {
    // 先取消KafkaMBean的注册
    unregister(mbean);
    try {
        // 调用MBeanServer的registerMBean()方法重新注册KafkaMBean
        ManagementFactory.getPlatformMBeanServer().registerMBean(mbean, mbean.name());
    } catch (JMException e) {
        throw new KafkaException("Error registering mbean " + mbean.name(), e);
    }
}

private String getMBeanName(MetricName metricName) {
    StringBuilder mBeanName = new StringBuilder();
    mBeanName.append(prefix); // 第一部分是前缀，服务端默认是kafka.server
    mBeanName.append(":type="); // 第二部分是MetricName中group字段
    mBeanName.append(metricName.group());
    // 第三部分由MetricName中的tags集合构成
    for (Map.Entry<String, String> entry : metricName.tags().entrySet()) {
        if (entry.getKey().length() <= 0 || entry.getValue().length() <= 0)
            continue;
        mBeanName.append(",");
        mBeanName.append(entry.getKey());
        mBeanName.append("=");
        mBeanName.append(entry.getValue());
    }
    return mBeanName.toString();
}

// 实现对brokerStateInfo集合的管理
private def addNewBroker(broker: Broker) {
  ...
  // 构建网络通信组件
  val networkClient = {
    // 创建ChannelBuilder对象
    val channelBuilder = ChannelBuilders.create(
      config.interBrokerSecurityProtocol,
      Mode.CLIENT,
      LoginType.SERVER,
      config.values,
      config.saslMechanismInterBrokerProtocol,
      config.saslInterBrokerHandshakeRequestEnable
    )
    // 创建Selector对象
    val selector = new Selector(
      NetworkReceive.UNLIMITED,
      config.connectionsMaxIdleMs,
      metrics, // Metrics对象，由KafkaServer启动时创建
      time,
      "controller-channel",
      Map("broker-id" -> broker.id.toString).asJava,
      false,
      channelBuilder
    )

    // 创建NetworkClient对象
    new NetworkClient(
      selector,
      new ManualMetadataUpdater(Seq(brokerNode).asJava),
      config.brokerId.toString,
      1,
      0,
      Selectable.USE_DEFAULT_BUFFER_SIZE,
      Selectable.USE_DEFAULT_BUFFER_SIZE,
      config.requestTimeoutMs,
      time
    )
  }
  ...
}

class KafkaServer(val config: KafkaConfig, time: Time = SystemTime, threadNamePrefix: Option[String] = None) extends Logging with KafkaMetricsGroup {
  ...
  // 用于构成MBean名称的一部分
  private val jmxPrefix: String = "kafka.server"

  // 读取配置的MetricReporter类型，metric.reporters，默认为空
  private val reporters: java.util.List[MetricsReporter] = config.metricReporterClasses
  // 默认会创建并添加JmxReporter对象
  reporters.add(new JmxReporter(jmxPrefix))
  ...
  // Metrics监控
  var metrics: Metrics = null
  // 创建MetricConfig对象
  private val metricConfig: MetricConfig = new MetricConfig()
    // 设置Sample个数，默认为2，metrics.num.samples
    .samples(config.metricNumSamples)
    // 设置Sample的时间，默认为30000，即300秒，metrics.sample.window.ms
    .timeWindow(config.metricSampleWindowMs, TimeUnit.MILLISECONDS)
  ...
  def startup() {
    ...
    metrics = new Metrics(metricConfig, reporters, kafkaMetricsTime, true)
    ...
  }
  ...
}

private class SelectorMetrics {
    private final Metrics metrics;
    // 监控连接关闭，使用Rate记录每秒连接关闭数
    public final Sensor connectionClosed;
    // 监控连接创建，使用Rate记录每秒连接创建数
    public final Sensor connectionCreated;
    // 监控网络操作数，使用Rate记录每秒钟所有连接上执行的读写操作总数
    public final Sensor bytesTransferred;
    /**
     * 监控发送请求的相关指标，此Sensor是bytesTransferred的子Sensor。
     * 其中使用Rate记录Controller每秒钟发送的字节数和请求数，
     * 使用Avg记录发送请求的平均大小，使用Max记录发送请求的最大长度。
     */
    public final Sensor bytesSent;
    /**
     * 监控接收请求的相关指标，此Sensor是byteTransferred的子Sensor。
     * 使用Rate记录Controller每秒钟接收的字节数和请求数。
     */
    public final Sensor bytesReceived;
    /**
     * 监控select()方法的相关指标。
     * 使用Rate记录每秒钟调用select()方法的次数，
     * 使用Avg记录调用select()方法阻塞的平均值，
     * 使用Rate记录调用select()方法阻塞时间占总时间的比例。
     */
    public final Sensor selectTime;
    /**
     * 监控I/O耗时的相关指标。
     * 使用Avg记录I/O操作的平均耗时，使用Rate执行I/O操作占总时间的比例。
     */
    public final Sensor ioTime;

    /**
     * Names of metrics that are not registered through sensors
     * 保存了直接向Metrics注册的Measurable对象，这些Measurable对象没有注册到Sensor中。
     * */
    private final List<MetricName> topLevelMetricNames = new ArrayList<>();
    // 保存了上面全部的Sensor对象
    private final List<Sensor> sensors = new ArrayList<>();
    ...
}

public SelectorMetrics(Metrics metrics) {
    this.metrics = metrics;
    // 此处得到的metricGrpName的值为controller-channel-metrics，下面所有Sensor都是用此值作为group部分
    String metricGrpName = metricGrpPrefix + "-metrics";
    StringBuilder tagsSuffix = new StringBuilder();

    /**
     * 将tags集合组装成字符串，假设连接的BrokerId为1，此值为broker-1，
     * 下面所有Sensor都会将此值作为其name的一部分
     */
    for (Map.Entry<String, String> tag: metricTags.entrySet()) {
        tagsSuffix.append(tag.getKey());
        tagsSuffix.append("-");
        tagsSuffix.append(tag.getValue());
    }

    // >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>

    // 创建Sensor对象
    this.connectionClosed = sensor("connections-closed:" + tagsSuffix.toString());
    // 创建MetricName
    MetricName metricName = metrics.metricName("connection-close-rate", metricGrpName, "Connections closed per second in the window.", metricTags);
    // 记录每秒连接的关闭数
    this.connectionClosed.add(metricName, new Rate());

    // >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>

    // 创建Sensor对象
    this.connectionCreated = sensor("connections-created:" + tagsSuffix.toString());

    // 记录每秒连接的创建数
    metricName = metrics.metricName("connection-creation-rate", metricGrpName, "New connections established per second in the window.", metricTags);
    this.connectionCreated.add(metricName, new Rate());

    // >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>

    // 创建Sensor对象
    this.bytesTransferred = sensor("bytes-sent-received:" + tagsSuffix.toString());

    // 记录所有连接每秒执行的读写总数
    metricName = metrics.metricName("network-io-rate", metricGrpName, "The average number of network operations (reads or writes) on all connections per second.", metricTags);
    bytesTransferred.add(metricName, new Rate(new Count()));

    // >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>

    // 创建Sensor对象，这里会将bytesTransferred作为付Sensor
    this.bytesSent = sensor("bytes-sent:" + tagsSuffix.toString(), bytesTransferred);

    // 记录所有连接每秒发送的总字节数
    metricName = metrics.metricName("outgoing-byte-rate", metricGrpName, "The average number of outgoing bytes sent per second to all servers.", metricTags);
    this.bytesSent.add(metricName, new Rate());

    // 记录平均每秒发送的总请求数
    metricName = metrics.metricName("request-rate", metricGrpName, "The average number of requests sent per second.", metricTags);
    this.bytesSent.add(metricName, new Rate(new Count()));

    // 记录请求的平均大小
    metricName = metrics.metricName("request-size-avg", metricGrpName, "The average size of all requests in the window..", metricTags);
    this.bytesSent.add(metricName, new Avg());

    // 记录请求的最大长度
    metricName = metrics.metricName("request-size-max", metricGrpName, "The maximum size of any request sent in the window.", metricTags);
    this.bytesSent.add(metricName, new Max());

    // >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>

    // 创建Sensor对象，这里会将bytesTransferred作为付Sensor
    this.bytesReceived = sensor("bytes-received:" + tagsSuffix.toString(), bytesTransferred);

    // 记录每秒收到的字节数
    metricName = metrics.metricName("incoming-byte-rate", metricGrpName, "Bytes/second read off all sockets", metricTags);
    this.bytesReceived.add(metricName, new Rate());

    // 记录每秒收到的请求数
    metricName = metrics.metricName("response-rate", metricGrpName, "Responses received sent per second.", metricTags);
    this.bytesReceived.add(metricName, new Rate(new Count()));

    // >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>

    // 创建Sensor对象
    this.selectTime = sensor("select-time:" + tagsSuffix.toString());

    // 记录每秒调用select()方法次数
    metricName = metrics.metricName("select-rate", metricGrpName, "Number of times the I/O layer checked for new I/O to perform per second", metricTags);
    this.selectTime.add(metricName, new Rate(new Count()));

    // 记录select()方法的平均阻塞时间
    metricName = metrics.metricName("io-wait-time-ns-avg", metricGrpName, "The average length of time the I/O thread spent waiting for a socket ready for reads or writes in nanoseconds.", metricTags);
    this.selectTime.add(metricName, new Avg());

    // 记录调用select()方法阻塞时间占总时间的比例
    metricName = metrics.metricName("io-wait-ratio", metricGrpName, "The fraction of time the I/O thread spent waiting.", metricTags);
    this.selectTime.add(metricName, new Rate(TimeUnit.NANOSECONDS));

    // >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>

    // 创建Sensor对象
    this.ioTime = sensor("io-time:" + tagsSuffix.toString());

    // 记录I/O的平均时长
    metricName = metrics.metricName("io-time-ns-avg", metricGrpName, "The average length of time for I/O per select call in nanoseconds.", metricTags);
    this.ioTime.add(metricName, new Avg());

    // 记录执行I/O时间占总时间的比例
    metricName = metrics.metricName("io-ratio", metricGrpName, "The fraction of time the I/O thread spent doing I/O", metricTags);
    this.ioTime.add(metricName, new Rate(TimeUnit.NANOSECONDS));

    // 记录连接数
    metricName = metrics.metricName("connection-count", metricGrpName, "The current number of active connections.", metricTags);
    topLevelMetricNames.add(metricName);

    // 直接向Metrics添加匿名的Measurale，用来记录连接数
    this.metrics.addMetric(metricName, new Measurable() {
        public double measure(MetricConfig config, long now) {
            return channels.size();
        }
    });
}

/**
 * Begin closing this connection
 */
private void close(KafkaChannel channel) {
    try {
        // 关闭连接
        channel.close();
    } catch (IOException e) {
        log.error("Exception closing connection to node {}:", channel.id(), e);
    }
    this.stagedReceives.remove(channel);
    this.channels.remove(channel.id());
    this.lruConnections.remove(channel.id());
    // 调用connectionClosed的record()，记录连接关闭数
    this.sensors.connectionClosed.record();
}

@Override
public void poll(long timeout) throws IOException {
    ...
    /* check ready keys */
    // 记录select()方法的起始时间
    long startSelect = time.nanoseconds();
    // 调用select()方法，等待事件发生
    int readyKeys = select(timeout);
    // 记录select()方法的结束时间
    long endSelect = time.nanoseconds();
    currentTimeNanos = endSelect;
    // 调用selectTime的record()方法，记录select()方法的阻塞时间
    this.sensors.selectTime.record(endSelect - startSelect, time.milliseconds());

    // 调用pollSelectionKeys()方法进行I/O操作处理
    if (readyKeys > 0 || !immediatelyConnectedKeys.isEmpty()) {
        // 处理IO事件
        pollSelectionKeys(this.nioSelector.selectedKeys(), false);
        pollSelectionKeys(immediatelyConnectedKeys, true);
    }

    // 将statedReceives复制到completedReceives集合中
    addToCompletedReceives();

    // 记录I/O操作的结束时间
    long endIo = time.nanoseconds();
    // 调用ioTime的record()方法记录I/O操作的耗时
    this.sensors.ioTime.record(endIo - endSelect, time.milliseconds());
    ...
}

// 处理IO操作
private void pollSelectionKeys(Iterable<SelectionKey> selectionKeys, boolean isImmediatelyConnected) {
    ...
    /* complete any connections that have finished their handshake (either normally or immediately) */
    // connect()方法返回true或OP_CONNECTION事件的处理
    if (isImmediatelyConnected || key.isConnectable()) {
        // finishConnect方法会先检测sockChannel是否建立完成，建立后，会取消对OP_CONNECT事件关注，开始关注OP_READ事件
        if (channel.finishConnect()) { // 检测是否已经成功连接
            // 将连接的ID添加到connected集合中
            this.connected.add(channel.id());
            // 调用connectionCreated的record()方法，记录连接创建数
            this.sensors.connectionCreated.record();
        } else
            // 连接未完成，跳过对该Channel的后续处理
            continue;
    }
    ...
    /* if channel is ready write to any sockets that have space in their buffer and for which we have data */
    if (channel.ready() && key.isWritable()) {
        // Channel可写，处理OP_WRITE事件
        Send send = channel.write();
        /**
          * channel.write()将KafkaChannel中保存的send字段发送出去，
          * 如果发送成功就会返回send，然后将其添加到completedSends集合，等待后续处理
          * 如果发送未完成会返回null
          */
        if (send != null) { // 成功发送一个完整的请求
            // 添加到completedSends集合等待后续处理
            this.completedSends.add(send);
            // 调用bytesSent的record()方法进行记录
            this.sensors.recordBytesSent(channel.id(), send.size());
        }
    }
    ...
}

/**
 * checks if there are any staged receives and adds to completedReceives
 */
private void addToCompletedReceives() {
    ...
    this.completedReceives.add(networkReceive);
    // 调用bytesReceived的record()方法进行记录
    this.sensors.recordBytesReceived(channel.id(), networkReceive.payload().limit());
    ...
}