import org.apache.storm.metric.api.ReducedMetric; //导入依赖的package包/类
private void initializeAndRegisterAllMetrics(TopologyContext context, int timeBucketSize) {
    this.failureMetric                  = new CombinedMetric(new MaxMetric());
    this.successCountMetric             = new CountMetric();
    this.sidelineCountMetric            = new CountMetric();
    this.internalBufferSize             = new CombinedMetric(new MaxMetric());
    this.pendingMessageSize             = new CombinedMetric(new MaxMetric());
    this.currentBinLogFileNumber        = new CombinedMetric(new MaxMetric());
    this.currentBinLogFilePosition      = new CombinedMetric(new MaxMetric());
    this.txEventProcessTime             = new ReducedMetric(new MeanReducer());
    this.txEventFailTimeInTopology      = new ReducedMetric(new MeanReducer());

    context.registerMetric(SpoutConstants.METRIC_FAILURECOUNT, this.failureMetric, timeBucketSize);
    context.registerMetric(SpoutConstants.METRIC_SUCCESSCOUNT, this.successCountMetric, timeBucketSize);
    context.registerMetric(SpoutConstants.METRIC_SIDELINECOUNT, this.sidelineCountMetric, timeBucketSize);
    context.registerMetric(SpoutConstants.METRIC_BUFFER_SIZE, this.internalBufferSize, timeBucketSize);
    context.registerMetric(SpoutConstants.METRIC_PENDING_MESSAGES, this.pendingMessageSize, timeBucketSize);
    context.registerMetric(SpoutConstants.METRIC_TXPROCESSTIME, this.txEventProcessTime, timeBucketSize);
    context.registerMetric(SpoutConstants.METRIC_BINLOG_FILE_NUM, this.currentBinLogFileNumber, timeBucketSize);
    context.registerMetric(SpoutConstants.METRIC_BIN_LOG_FILE_POS, this.currentBinLogFilePosition, timeBucketSize);
    context.registerMetric(SpoutConstants.METRIC_FAIL_MSG_IN_TOPOLOGY, this.txEventFailTimeInTopology, timeBucketSize);

}
 

import org.apache.storm.metric.api.ReducedMetric; //导入依赖的package包/类
@Override
@SuppressWarnings({ "rawtypes", "unchecked" })
public void prepare(Map stormConf, TopologyContext context,
        OutputCollector collector) {
    super.prepare(stormConf, context, collector);
    sniffWhenNoSMKey = ConfUtils.getBoolean(stormConf,
            "sitemap.sniffContent", false);
    filterHoursSinceModified = ConfUtils.getInt(stormConf,
            "sitemap.filter.hours.since.modified", -1);
    parseFilters = ParseFilters.fromConf(stormConf);
    maxOffsetGuess = ConfUtils.getInt(stormConf, "sitemap.offset.guess",
            300);
    averagedMetrics = context.registerMetric(
            "sitemap_average_processing_time", new ReducedMetric(
                    new MeanReducer()), 30);
}
 

import org.apache.storm.metric.api.ReducedMetric; //导入依赖的package包/类
private ReducedMetric registerAveragingMetric(String name, TopologyContext context) {
    Number interval = metricsIntervalMapping.getOrDefault(name,
                                                          metricsIntervalMapping.get(DEFAULT_BUILT_IN_METRICS_INTERVAL_KEY));
    log.info("Registered {} with interval {}", name, interval);
    return context.registerMetric(name, new ReducedMetric(new MeanReducer()), interval.intValue());
}
 

import org.apache.storm.metric.api.ReducedMetric; //导入依赖的package包/类
@SuppressWarnings("rawtypes")
public ReducedMetric registerMetric(String name, IReducer reducer, int timeBucketSizeInSecs) {
  return registerMetric(name, new ReducedMetric(reducer), timeBucketSizeInSecs);
}
 

import org.apache.storm.metric.api.ReducedMetric; //导入依赖的package包/类
@SuppressWarnings("rawtypes")
ReducedMetric registerMetric(String name, IReducer reducer, int timeBucketSizeInSecs);
 

import org.apache.storm.metric.api.ReducedMetric; //导入依赖的package包/类
/**
 * Not supported by Flink.
 *
 * @throws UnsupportedOperationException
 * 		at every invocation
 */
@SuppressWarnings("rawtypes")
@Override
public ReducedMetric registerMetric(final String name, final IReducer combiner, final int timeBucketSizeInSecs) {
	throw new UnsupportedOperationException("Metrics are not supported by Flink");
}