在spout或者bolt的prepare方法中注册metrics
public void prepare(@SuppressWarnings("rawtypes") Map stormConf, TopologyContext context, OutputCollector collector) { this.collector = collector; barMetric = new CountMetric(); context.registerMetric("bar", this.barMetric, 3); }
在恰当的时候更新metrics
public void execute(Tuple input) { if (input.getSourceComponent().equals("bar")) { barMetric.incr(); } collector.ack(input); }
启动拓扑的时候注册metrics的consumer
conf.registerMetricsConsumer(MetricsConsumer.class);