import org.apache.storm.metric.api.IMetric; //导入依赖的package包/类
@Override
public void prepare(Map stormConf, TopologyContext context,
        OutputCollector collector) {
    _collector = collector;
    indexName = ConfUtils.getString(stormConf, ESStatusIndexNameParamName,
            "status");
    docType = ConfUtils.getString(stormConf, ESStatusDocTypeParamName,
            "doc");
    try {
        connection = ElasticSearchConnection.getConnection(stormConf,
                ESBoltType);
    } catch (Exception e1) {
        LOG.error("Can't connect to ElasticSearch", e1);
        throw new RuntimeException(e1);
    }

    context.registerMetric("status.count", new IMetric() {
        @Override
        public Object getValueAndReset() {
            return latestStatusCounts;
        }
    }, freqStats);
}
 

import org.apache.storm.metric.api.IMetric; //导入依赖的package包/类
@Override
public <T extends IMetric> T registerMetric(String name, T metric, int timeBucketSizeInSecs) {
    Map<String, IMetric> metrics = registeredMetrics.getOrDefault(timeBucketSizeInSecs, new HashMap<>());
    metrics.put(name, metric);
    registeredMetrics.putIfAbsent(timeBucketSizeInSecs, metrics);
    return metric;
}
 

import org.apache.storm.metric.api.IMetric; //导入依赖的package包/类
@Override
public IMetric getRegisteredMetricByName(String name) {
    Optional<Map.Entry<String, IMetric>> metric = registeredMetrics.values().stream()
                                                                   .flatMap(m -> m.entrySet().stream())
                                                                   .filter(e -> e.getKey().equals(name))
                                                                   .findFirst();
    return metric.isPresent() ? metric.get().getValue() : null;
}
 

import org.apache.storm.metric.api.IMetric; //导入依赖的package包/类
public IMetric getRegisteredMetricInTimeBucket(Integer timeBucket, String name) {
    Optional<Map.Entry<String, IMetric>> metric = registeredMetrics.getOrDefault(timeBucket, Collections.emptyMap())
                                                                   .entrySet().stream()
                                                                   .filter(e -> e.getKey().equals(name))
                                                                   .findFirst();
    return metric.isPresent() ? metric.get().getValue() : null;
}
 

import org.apache.storm.metric.api.IMetric; //导入依赖的package包/类
@Test(expected = UnsupportedOperationException.class)
public void testRegisteredMetric3() {
	new FlinkTopologyContext(new StormTopology(new HashMap<String, SpoutSpec>(),
			new HashMap<String, Bolt>(), new HashMap<String, StateSpoutSpec>()), null, null,
			null, null, null, null, null, null, null, null, null, null, null, null, null)
	.registerMetric(null, (IMetric) null, 0);
}
 

import org.apache.storm.metric.api.IMetric; //导入依赖的package包/类
@SuppressWarnings({ "unchecked", "rawtypes" })
@Override
public void prepare(Map stormConf, TopologyContext context,
        OutputCollector collector) {
    _collector = collector;

    scheduler = Scheduler.getInstance(stormConf);

    mdTransfer = MetadataTransfer.getInstance(stormConf);

    useCache = ConfUtils.getBoolean(stormConf, useCacheParamName, true);

    if (useCache) {
        String spec = ConfUtils.getString(stormConf, cacheConfigParamName);
        cache = CacheBuilder.from(spec).build();

        context.registerMetric("cache", new IMetric() {
            @Override
            public Object getValueAndReset() {
                Map<String, Long> statsMap = new HashMap<>();
                statsMap.put("hits", cacheHits);
                statsMap.put("misses", cacheMisses);
                statsMap.put("size", cache.size());
                cacheHits = 0;
                cacheMisses = 0;
                return statsMap;
            }
        }, 30);
    }

    maxFetchErrors = ConfUtils
            .getInt(stormConf, maxFetchErrorsParamName, 3);
}
 

import org.apache.storm.metric.api.IMetric; //导入依赖的package包/类
@Override
public void open(@SuppressWarnings("rawtypes") Map conf,
        TopologyContext context, SpoutOutputCollector collector) {
    _collector = collector;

    // check that there is only one instance of it
    int totalTasks = context
            .getComponentTasks(context.getThisComponentId()).size();
    if (totalTasks > 1) {
        throw new RuntimeException(
                "Can't have more than one instance of the MemorySpout");
    }

    Date now = new Date();
    for (String u : startingURLs) {
        LOG.debug("About to deserialize {} ", u);
        List<Object> tuple = scheme.deserialize(ByteBuffer.wrap(u
                .getBytes(StandardCharsets.UTF_8)));
        add((String) tuple.get(0), (Metadata) tuple.get(1), now);
    }

    context.registerMetric("queue_size", new IMetric() {
        @Override
        public Object getValueAndReset() {
            return queue.size();
        }
    }, 10);

}
 

import org.apache.storm.metric.api.IMetric; //导入依赖的package包/类
public static TopologyContext getMockedTopologyContext() {
    TopologyContext context = mock(TopologyContext.class);
    when(context.registerMetric(anyString(), any(IMetric.class), anyInt()))
            .thenAnswer(new Answer<IMetric>() {

                @Override
                public IMetric answer(InvocationOnMock invocation)
                        throws Throwable {
                    return invocation.getArgumentAt(1, IMetric.class);
                }
            });
    return context;
}
 

import org.apache.storm.metric.api.IMetric; //导入依赖的package包/类
@Override
public void prepare(Map stormConf, TopologyContext context,
        OutputCollector collector) {

    super.prepare(stormConf, context, collector);

    indexName = ConfUtils.getString(stormConf,
            StatusUpdaterBolt.ESStatusIndexNameParamName, "status");
    docType = ConfUtils.getString(stormConf,
            StatusUpdaterBolt.ESStatusDocTypeParamName, "status");

    doRouting = ConfUtils.getBoolean(stormConf,
            StatusUpdaterBolt.ESStatusRoutingParamName, false);

    if (doRouting) {
        partitioner = new URLPartitioner();
        partitioner.configure(stormConf);
        fieldNameForRoutingKey = ConfUtils.getString(stormConf,
                StatusUpdaterBolt.ESStatusRoutingFieldParamName);
        if (StringUtils.isNotBlank(fieldNameForRoutingKey)) {
            if (fieldNameForRoutingKey.startsWith("metadata.")) {
                routingFieldNameInMetadata = true;
                fieldNameForRoutingKey = fieldNameForRoutingKey
                        .substring("metadata.".length());
            }
            // periods are not allowed in ES2 - replace with %2E
            fieldNameForRoutingKey = fieldNameForRoutingKey.replaceAll(
                    "\\.", "%2E");
        }
    }

    waitAck = CacheBuilder.newBuilder()
            .expireAfterWrite(60, TimeUnit.SECONDS).removalListener(this)
            .build();

    // create gauge for waitAck
    context.registerMetric("waitAck", new IMetric() {
        @Override
        public Object getValueAndReset() {
            return waitAck.size();
        }
    }, 30);

    try {
        connection = ElasticSearchConnection.getConnection(stormConf,
                ESBoltType, this);
    } catch (Exception e1) {
        LOG.error("Can't connect to ElasticSearch", e1);
        throw new RuntimeException(e1);
    }

    this.eventCounter = context.registerMetric("counters",
            new MultiCountMetric(), 30);
}
 

import org.apache.storm.metric.api.IMetric; //导入依赖的package包/类
@Override
public <T extends IMetric> T registerMetric(String name, T metric, int timeBucketSizeInSecs) {
    mockRegisteredMetrics.put(name, metric);
    return metric;
}
 

import org.apache.storm.metric.api.IMetric; //导入依赖的package包/类
@Override
public IMetric getRegisteredMetricByName(String name) {
    return mockRegisteredMetrics.get(name);
}
 

import org.apache.storm.metric.api.IMetric; //导入依赖的package包/类
private Number fetchResult(IMetric metric) {
    return metric == null ? null : (Number) metric.getValueAndReset();
}
 

import org.apache.storm.metric.api.IMetric; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public <T extends IMetric> T registerMetric(String name, T metric, int timeBucketSizeInSecs) {
  MetricDelegate d = new MetricDelegate(metric);
  delegate.registerMetric(name, d, timeBucketSizeInSecs);
  return metric;
}
 

import org.apache.storm.metric.api.IMetric; //导入依赖的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;
}
 

import org.apache.storm.metric.api.IMetric; //导入依赖的package包/类
/**
 * Not supported by Flink.
 *
 * @throws UnsupportedOperationException
 * 		at every invocation
 */
@Override
public IMetric getRegisteredMetricByName(final String name) {
	throw new UnsupportedOperationException("Metrics are not supported by Flink");

}
 

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

import org.apache.storm.metric.api.IMetric; //导入依赖的package包/类
<T extends IMetric> T registerMetric(String name, T metric, int timeBucketSizeInSecs);