13518219792
今天就跟大家聊聊有关如何实现 LoggingMetricsConsumer将指标值输出到metric.log日志文件，可能很多人都不太了解，为了让大家更加了解，小编给大家总结了以下内容，希望大家根据这篇文章可以有所收获。
公司主营业务：成都做网站、成都网站制作、移动网站开发等业务。帮助企业客户真正实现互联网宣传，提高企业的竞争能力。成都创新互联是一支青春激扬、勤奋敬业、活力青春激扬、勤奋敬业、活力澎湃、和谐高效的团队。公司秉承以“开放、自由、严谨、自律”为核心的企业文化，感谢他们对我们的高要求，感谢他们从不同领域给我们带来的挑战，让我们激情的团队有机会用头脑与智慧不断的给客户带来惊喜。成都创新互联推出濂溪免费做网站回馈大家。
前提说明：
storm从0.9.0开始，增加了指标统计框架，用来收集应用程序的特定指标，并将其输出到外部系统。
一般来说，您只需要去实现 LoggingMetricsConsumer，统计将指标值输出到metric.log日志文件之中。
当然，您也可以自定义一个监听的类：只需要去实现IMetricsConsumer接口就可以了。这些类可以在代码里注册（registerMetricsConsumer），也可以在 storm.yaml配置文件中注册：
package com.digitalpebble.storm.crawler;
import backtype.storm.Config;
import backtype.storm.metric.MetricsConsumerBolt;
import backtype.storm.metric.api.IMetricsConsumer;
import backtype.storm.task.IErrorReporter;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;
import backtype.storm.utils.Utils;
import com.google.common.base.Joiner;
import com.google.common.base.Supplier;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSortedMap;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.map.ObjectWriter;
import org.mortbay.jetty.Server;
import org.mortbay.jetty.servlet.Context;
import org.mortbay.jetty.servlet.ServletHolder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
/**
* @author Enno Shioji (enno.shioji@peerindex.com)
*/
public class DebugMetricConsumer implements IMetricsConsumer {
private static final Logger log = LoggerFactory
.getLogger(DebugMetricConsumer.class);
private IErrorReporter errorReporter;
private Server server;
// Make visible to servlet threads
private volatile TopologyContext context;
private volatile ConcurrentMap metrics;
private volatile ConcurrentMap> metrics_metadata;
public void prepare(Map stormConf, Object registrationArgument,
TopologyContext context, IErrorReporter errorReporter) {
this.context = context;
this.errorReporter = errorReporter;
this.metrics = new ConcurrentHashMap();
this.metrics_metadata = new ConcurrentHashMap>();
try {
// TODO Config file not tested
final String PORT_CONFIG_STRING = "topology.metrics.consumers.debug.servlet.port";
Integer port = (Integer) stormConf.get(PORT_CONFIG_STRING);
if (port == null) {
log.warn("Metrics debug servlet's port not specified, defaulting to 7070. You can specify it via "
+ PORT_CONFIG_STRING + " in storm.yaml");
port = 7070;
}
server = startServlet(port);
} catch (Exception e) {
log.error("Failed to start metrics server", e);
throw new AssertionError(e);
}
}
private static final Joiner ON_COLONS = Joiner.on("::");
public void handleDataPoints(TaskInfo taskInfo,
Collection dataPoints) {
// In order
String componentId = taskInfo.srcComponentId;
Integer taskId = taskInfo.srcTaskId;
Integer updateInterval = taskInfo.updateIntervalSecs;
Long timestamp = taskInfo.timestamp;
for (DataPoint point : dataPoints) {
String metric_name = point.name;
try {
Map metric = (Map) point.value;
for (Map.Entry entry : metric.entrySet()) {
String metricId = ON_COLONS.join(componentId, taskId,
metric_name, entry.getKey());
Number val = entry.getValue();
metrics.put(metricId, val);
metrics_metadata.put(metricId, ImmutableMap
. of("updateInterval",
updateInterval, "lastreported", timestamp));
}
} catch (RuntimeException e) {
// One can easily send something else than a Map
// down the __metrics stream and make this part break.
// If you ask me either the message should carry type
// information or there should be different stream per message
// type
// This is one of the reasons why I want to write a further
// abstraction on this facility
errorReporter.reportError(e);
metrics_metadata
.putIfAbsent("ERROR_METRIC_CONSUMER_"
+ e.getClass().getSimpleName(), ImmutableMap
.of("offending_message_sample", point.value));
}
}
}
private static final ObjectMapper OM = new ObjectMapper();
private Server startServlet(int serverPort) throws Exception {
// Setup HTTP server
Server server = new Server(serverPort);
Context root = new Context(server, "/");
server.start();
HttpServlet servlet = new HttpServlet() {
@Override
protected void doGet(HttpServletRequest req,
HttpServletResponse resp) throws ServletException,
IOException {
SortedMap metrics = ImmutableSortedMap
.copyOf(DebugMetricConsumer.this.metrics);
SortedMap> metrics_metadata = ImmutableSortedMap
.copyOf(DebugMetricConsumer.this.metrics_metadata);
Map toplevel = ImmutableMap
.of("retrieved",
new Date(),
// TODO this call fails with mysterious
// exception
// "java.lang.IllegalArgumentException: Could not find component common for __metrics"
// Mailing list suggests it's a library version
// issue but couldn't find anything suspicious
// Need to eventually investigate
// "sources",
// context.getThisSources().toString(),
"metrics", metrics, "metric_metadata",
metrics_metadata);
ObjectWriter prettyPrinter = OM
.writerWithDefaultPrettyPrinter();
prettyPrinter.writeValue(resp.getWriter(), toplevel);
}
};
root.addServlet(new ServletHolder(servlet), "/metrics");
log.info("Started metric server...");
return server;
}
public void cleanup() {
try {
server.stop();
} catch (Exception e) {
throw new AssertionError(e);
}
}
} 看完上述内容，你们对如何实现 LoggingMetricsConsumer将指标值输出到metric.log日志文件有进一步的了解吗？如果还想了解更多知识或者相关内容，请关注创新互联行业资讯频道，感谢大家的支持。
