摘要：免費領(lǐng)取驗證碼內(nèi)容安全短信發(fā)送直播點播體驗包及云服務(wù)器等套餐更多網(wǎng)易技術(shù)產(chǎn)品運營經(jīng)驗分享請訪問網(wǎng)易云社區(qū)。文章來源網(wǎng)易云社區(qū)

本文由作者林洋港授權(quán)網(wǎng)易云社區(qū)發(fā)布。

作為服務(wù)端程序，我們總是需要向外界報告一些統(tǒng)計數(shù)據(jù)，以助于了解系統(tǒng)的運行情況，比如某個接口的調(diào)用時間、系統(tǒng)處理的請求數(shù)等等。當(dāng)我們的程序以Storm Topology的形式運行時同樣需要輸出這些統(tǒng)計數(shù)據(jù)。Storm為我們提供了Metric接口，可以方便的把一些統(tǒng)計指標(biāo)輸出到指定的地方。Storm Metric的統(tǒng)計方式為每隔指定的時間間隔輸出統(tǒng)計內(nèi)容。本文首先介紹Storm Metric相關(guān)的接口以及它們之間的關(guān)系，然后以實際應(yīng)用中的一個例子來說明如何使用Metric接口。本文使用的Storm版本為0.9.1-incubating。

IMetric是Storm用于保存統(tǒng)計數(shù)據(jù)的接口

public interface IMetric {

public Object getValueAndReset();

}

接口只有一個getValueAndReset方法，當(dāng)需要輸出統(tǒng)計內(nèi)容時，Storm就會調(diào)用這個方法。值得注意的是getValueAndReset方法返回的是Object類型，這為統(tǒng)計內(nèi)容的形式提供了靈活性，我們可以返回任意的類型作為統(tǒng)計信息，這一點在后面的例子中我們會再提到。另一個引起我們注意的地方是IMetric接口并沒有聲明更新統(tǒng)計數(shù)據(jù)的方法，這樣當(dāng)我們實現(xiàn)IMetric接口的時候就更加靈活了——參數(shù)類型、參數(shù)個數(shù)都沒有限制。Storm自身提供了6個IMetric實現(xiàn)：AssignableMetric、CombinedMetric、CountMetric、MultiCountMetric、ReducedMetric、StateMetric。這里只介紹CountMetric和MultiCountMetric的使用方式，以印證前面說的IMetric接口統(tǒng)計數(shù)據(jù)更新方式的靈活性以及getValueAndReset返回Object類型的靈活性。CountMetric就是一個簡單的計數(shù)器，有兩個方法incr()和incrBy(long incrementBy)，其getValueAndReset方法返回一個long類型的值：

public Object getValueAndReset() {

    long ret = _value;

    _value = 0;

    return ret;

}

MultiCountMetric，顧名思義，就是多個指標(biāo)的計數(shù)器，維護(hù)著一個Map，只有一個方法CountMetric scope(String key)。因此MultiCountMetric的更新方式為MultiCountMetric.scope(key).incr()或MultiCountMetric.scope(key).incrBy(long incrementBy)。它的getValueAndReset返回的是一個Map：

public Object getValueAndReset() {

    Map ret = new HashMap();

    for(Map.Entry e : _value.entrySet()) {

        ret.put(e.getKey(), e.getValue().getValueAndReset());

    }

    return ret;

}

除了IMetric接口，還有另外一個接口IMetricsConsumer，它負(fù)責(zé)向外輸出統(tǒng)計信息，即把IMetric getValueAndReset方法返回的數(shù)據(jù)輸出到外面。IMetricsConsumer有三個方法

void prepare(Map stormConf, Object registrationArgument, TopologyContext context, IErrorReporter errorReporter);

void handleDataPoints(TaskInfo taskInfo, Collection dataPoints);

void cleanup();

其中prepare是初始化，cleanup是生命周期結(jié)束時的清理工作，handleDataPoints才是真正的統(tǒng)計信息輸出方法，taskInfo參數(shù)存儲當(dāng)前task的信息（host、port、component id、task id等等），dataPoints存儲的是IMetric返回的統(tǒng)計信息，可能是出于性能考慮，dataPoints是一個集合，包含了多個IMetric返回的數(shù)據(jù)。讓我們來具體看看DataPoint這個類：

public static class DataPoint {

    @Override

    public String toString() {

        return "[" + name + " = " + value + "]";

    }

    public String name; 

    public Object value;

}

name是IMetric注冊時的名字，value就是IMetric getValueAndReset返回的那個Object。

Storm只提供了一個IMetricsConsumer實現(xiàn)——LoggingMetricsConsumer。LoggingMetricsConsumer做的事情很簡單，就是把dataPoints輸出到日志文件metrics.log，下面是其handleDataPoints方法的部分代碼：

   for (DataPoint p : dataPoints) {

        sb.delete(header.length(), sb.length());

        sb.append(p.name)

            .append(padding).delete(header.length()+23,sb.length()).append("	")

            .append(p.value);

        LOG.info(sb.toString());

    }

可以看到它通過調(diào)用DataPoint的value的toString方法把統(tǒng)計信息輸出到日志里面的，所以如果你的IMetric實現(xiàn)返回的是自己定義的類型，記得重載toString()方法，讓統(tǒng)計信息以可讀的格式輸出。

到這里Storm的Metric接口和自帶的實現(xiàn)基本介紹完了，接下來我們來看看怎么使用Storm自帶的這些實現(xiàn)。首先，Storm默認(rèn)的配置是關(guān)掉Metric功能的，可以有兩種方式開啟Metric功能：

1）在storm.yaml里面配置，這種是集群級別的設(shè)置，個人不建議這么做，所以就不多介紹了

2）conf.registerMetricsConsumer(Class klass, long parallelismHint);這是topology級別的，klass是IMetricsConsumer的實現(xiàn)類，parallelismHint這個參數(shù)Storm代碼里面沒注釋我也沒深入看底層的實現(xiàn)，這里結(jié)合自己的實驗談?wù)勊囊饬x：topology是在1個或多個worker上面以多個task的方式跑的嘛，parallelismHint就是指定多少個并發(fā)來輸出統(tǒng)計信息。這里我也不知道parallelismHint指的是多個task、worker還是supervisor，反正parallelismHint=1的時候只在特定的一個supervisor下面的metrics.log有統(tǒng)計信息，parallelismHint>1時可能取決于worker的數(shù)量，我測試的時候由于是在多個supervisor上跑的，因此觀察到多個supervisor都有metrics.log的輸出。個人經(jīng)驗是parallelismHint設(shè)為1，這樣可以在一個supervisor下面的metrics.log就能看到所有task的統(tǒng)計信息。

由于我建議采用第二種方法，所以示例代碼為：

//客戶端注冊IMetricsConsumer

conf.registerMetricsConsumer(LoggingMetricsConsumer.class);

StormSubmitter.submitTopology(name, conf, builder.createTopology());

//我們假設(shè)要統(tǒng)計spout某段代碼的調(diào)用次數(shù)

//注冊IMetric

@Override

public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {

    ...

    metric=new CountMetric();

    context.registerMetric("spout time cost", metric, 60);  //因此DataPoint的name為spout time cost，60表示1分鐘統(tǒng)計一次

    ...

}

//更新統(tǒng)計數(shù)據(jù)

@Override

public void nextTuple() {

    if(...)...

    else{

        ...

        metric.incr();

    }

}

這樣就可以了，然后你就能在metrics.log看到統(tǒng)計數(shù)據(jù)了。

現(xiàn)在，假設(shè)我們的需求跟上面不太一樣：1）metrics.log只打印我們自己維護(hù)的統(tǒng)計信息，屏蔽__system、__fail-count這種系統(tǒng)自己的統(tǒng)計信息；2）不只統(tǒng)計代碼的調(diào)用次數(shù)，還要統(tǒng)計調(diào)用時間——最小時間、最大時間、平均時間。

第一點可以通過重載LoggingMetricsConsumer的方法來實現(xiàn)：

public class AppLoggingMetricsConsumer extends LoggingMetricsConsumer {

@Override

public void handleDataPoints(TaskInfo taskInfo, Collection dataPoints) {

    if (taskInfo.srcComponentId != null && taskInfo.srcComponentId.startsWith("__")) return;

    if (dataPoints == null || dataPoints.isEmpty()) return;

    List list = new ArrayList();

    for (DataPoint p : dataPoints) {

        if (p.name == null || p.name.startsWith("__")) continue;

        list.add(p);

    }

    if (list.isEmpty()) return;

    super.handleDataPoints(taskInfo, list);

}

}

第二點需要開發(fā)我們自己的IMetric接口實現(xiàn)類TimeCostMetric，以下是其主要代碼：

@Override

public Object getValueAndReset() {

    TimeCost timeCost=new TimeCost();

    timeCost.count=count;

    if(timeCost.count>0){

        timeCost.min=min;

        timeCost.max=max;

        timeCost.mean=all*1.0/timeCost.count;

    }

    init();

    return timeCost;

}

public void update(long time){

    count++;

    all+=time;

    if(min>time)min=time;

    if(max
public static class TimeCost implements Serializable{
    private static final long serialVersionUID = 8355726599226036228L;

    int count;

    long min;

    long max;

    double mean;

    

    public String toString(){

        return "count: "+count+", min: "+min+", max:"+max+", mean: "+mean;

    }

}

TimeCostMetric的getValueAndReset方法返回的是一個TimeCost 對象，日志中最終打印的就是其toString方法的內(nèi)容。然后把前面紅色部分的代碼改成下面的內(nèi)容：
① conf.registerMetricsConsumer(AppLoggingMetricsConsumer .class);
② metric=new TimeCostMetric();
  context.registerMetric("MQ spout time cost", metric, 60);

③ metric.incr();
再來看看metrics.log

本文中是直接把統(tǒng)計信息打到日志中，你也可以自己實現(xiàn)IMetricsConsumer接口，把統(tǒng)計信息保存到指定的地方，如數(shù)據(jù)庫、監(jiān)控平臺等等。
免費領(lǐng)取驗證碼、內(nèi)容安全、短信發(fā)送、直播點播體驗包及云服務(wù)器等套餐
更多網(wǎng)易技術(shù)、產(chǎn)品、運營經(jīng)驗分享請訪問網(wǎng)易云社區(qū)。
文章來源： 網(wǎng)易云社區(qū)