import org.apache.storm.metric.api.MeanReducer; //导入依赖的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.MeanReducer; //导入依赖的package包/类
@SuppressWarnings({ "rawtypes", "unchecked" })
@Override
public void prepare(Map stormConf, TopologyContext context,
        OutputCollector collector) {
    super.prepare(stormConf, context, collector);

    partitioner = new URLPartitioner();
    partitioner.configure(stormConf);

    this.averagedMetrics = context.registerMetric("SQLStatusUpdater",
            new MultiReducedMetric(new MeanReducer()), 10);

    this.eventCounter = context.registerMetric("counter",
            new MultiCountMetric(), 10);

    tableName = ConfUtils.getString(stormConf,
            Constants.MYSQL_TABLE_PARAM_NAME);

    try {
        connection = SQLUtil.getConnection(stormConf);
    } catch (SQLException ex) {
        LOG.error(ex.getMessage(), ex);
        throw new RuntimeException(ex);
    }

}
 

import org.apache.storm.metric.api.MeanReducer; //导入依赖的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.MeanReducer; //导入依赖的package包/类
@Override
public void open(final Map<String, Object> spoutConfig, final TopologyContext topologyContext) {
    // Load configuration items.

    // Determine our time bucket window, in seconds, defaulted to 60.
    int timeBucketSeconds = 60;
    if (spoutConfig.containsKey(SpoutConfig.METRICS_RECORDER_TIME_BUCKET)) {
        final Object timeBucketCfgValue = spoutConfig.get(SpoutConfig.METRICS_RECORDER_TIME_BUCKET);
        if (timeBucketCfgValue instanceof Number) {
            timeBucketSeconds = ((Number) timeBucketCfgValue).intValue();
        }
    }

    // Conditionally enable prefixing with taskId
    if (spoutConfig.containsKey(SpoutConfig.METRICS_RECORDER_ENABLE_TASK_ID_PREFIX)) {
        final Object taskIdCfgValue = spoutConfig.get(SpoutConfig.METRICS_RECORDER_ENABLE_TASK_ID_PREFIX);
        if (taskIdCfgValue instanceof Boolean && (Boolean) taskIdCfgValue) {
            this.metricPrefix = "task-" + topologyContext.getThisTaskIndex();
        }
    }

    // Log how we got configured.
    logger.info("Configured with time window of {} seconds and using taskId prefixes?: {}",
        timeBucketSeconds, Boolean.toString(metricPrefix.isEmpty()));

    // Register the top level metrics.
    assignedValues = topologyContext.registerMetric("GAUGES", new MultiAssignableMetric(), timeBucketSeconds);
    timers = topologyContext.registerMetric("TIMERS", new MultiReducedMetric(new MeanReducer()), timeBucketSeconds);
    counters = topologyContext.registerMetric("COUNTERS", new MultiCountMetric(), timeBucketSeconds);
}
 

import org.apache.storm.metric.api.MeanReducer; //导入依赖的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.MeanReducer; //导入依赖的package包/类
@SuppressWarnings({ "rawtypes", "unchecked" })
@Override
public void prepare(Map stormConf, TopologyContext context,
        OutputCollector collector) {
    super.prepare(stormConf, context, collector);
    this.conf = new Config();
    this.conf.putAll(stormConf);

    checkConfiguration();

    this.taskID = context.getThisTaskId();

    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss",
            Locale.ENGLISH);
    long start = System.currentTimeMillis();
    LOG.info("[Fetcher #{}] : starting at {}", taskID, sdf.format(start));

    // Register a "MultiCountMetric" to count different events in this bolt
    // Storm will emit the counts every n seconds to a special bolt via a
    // system stream
    // The data can be accessed by registering a "MetricConsumer" in the
    // topology

    int metricsTimeBucketSecs = ConfUtils.getInt(conf,
            "fetcher.metrics.time.bucket.secs", 10);

    this.eventCounter = context.registerMetric("fetcher_counter",
            new MultiCountMetric(), metricsTimeBucketSecs);

    this.averagedMetrics = context.registerMetric("fetcher_average",
            new MultiReducedMetric(new MeanReducer()),
            metricsTimeBucketSecs);

    this.perSecMetrics = context.registerMetric("fetcher_average_persec",
            new MultiReducedMetric(new PerSecondReducer()),
            metricsTimeBucketSecs);

    // create gauges
    context.registerMetric("activethreads", new IMetric() {
        @Override
        public Object getValueAndReset() {
            return activeThreads.get();
        }
    }, metricsTimeBucketSecs);

    context.registerMetric("throttler_size", new IMetric() {
        @Override
        public Object getValueAndReset() {
            return throttler.size();
        }
    }, metricsTimeBucketSecs);

    protocolFactory = new ProtocolFactory(conf);

    sitemapsAutoDiscovery = ConfUtils.getBoolean(stormConf,
            SITEMAP_DISCOVERY_PARAM_KEY, false);

    queueMode = ConfUtils.getString(conf, "fetcher.queue.mode",
            QUEUE_MODE_HOST);
    // check that the mode is known
    if (!queueMode.equals(QUEUE_MODE_IP)
            && !queueMode.equals(QUEUE_MODE_DOMAIN)
            && !queueMode.equals(QUEUE_MODE_HOST)) {
        LOG.error("Unknown partition mode : {} - forcing to byHost",
                queueMode);
        queueMode = QUEUE_MODE_HOST;
    }
    LOG.info("Using queue mode : {}", queueMode);

    this.crawlDelay = (long) (ConfUtils.getFloat(conf,
            "fetcher.server.delay", 1.0f) * 1000);

    this.maxCrawlDelay = (long) ConfUtils.getInt(conf,
            "fetcher.max.crawl.delay", 30) * 1000;
}