在上一篇文章中，我們創建了一個簡單的索引代碼，該代碼可以對ElasticSearch進行數千個并發請求。 監視系統性能的唯一方法是老式的日志記錄語句：

.window(Duration.ofSeconds(1)) .flatMap(Flux::count) .subscribe(winSize -> log.debug("Got {} responses in last second", winSize));

很好，但是在生產系統上，我們寧愿有一些集中的監視和圖表解決方案來收集各種指標。 一旦在數千個實例中擁有數百個不同的應用程序，這一點就變得尤為重要。 具有單個圖形儀表板，匯總??所有重要信息變得至關重要。 我們需要兩個組件來收集一些指標：

• 發布指標
• 收集并可視化它們

使用Dropwizard指標發布指標

在Spring Boot 2中， Dropwizard指標被千分尺取代。 本文使用前者，下一個將在實踐中顯示后者的解決方案。 為了利用Dropwizard指標，我們必須將MetricRegistry或特定指標注入我們的業務類別。

import com.codahale.metrics.Counter; import com.codahale.metrics.MetricRegistry; import com.codahale.metrics.Timer; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j;@Component @RequiredArgsConstructor class Indexer {private final PersonGenerator personGenerator;private final RestHighLevelClient client;private final Timer indexTimer;private final Counter indexConcurrent;private final Counter successes;private final Counter failures;public Indexer(PersonGenerator personGenerator, RestHighLevelClient client, MetricRegistry metricRegistry) {this.personGenerator = personGenerator;this.client = client;this.indexTimer = metricRegistry.timer(name("es", "index"));this.indexConcurrent = metricRegistry.counter(name("es", "concurrent"));this.successes = metricRegistry.counter(name("es", "successes"));this.failures = metricRegistry.counter(name("es", "failures"));}private Flux<IndexResponse> index(int count, int concurrency) {//....}}

這么多樣板，以添加一些指標！

• indexTimer測量索引請求的時間分布（平均值，中位數和各種百分位數）
• indexConcurrent度量當前有多少個待處理的請求（已發送請求，尚未收到響應）； 指標隨時間上升和下降
• success和failures計算相應的成功索引請求和失敗索引請求的總數

我們將在一秒鐘內刪除樣板，但首先，讓我們看一下它在我們的業務代碼中的作用：

private Mono<IndexResponse> indexDocSwallowErrors(Doc doc) {return indexDoc(doc).doOnSuccess(response -> successes.inc()).doOnError(e -> log.error("Unable to index {}", doc, e)).doOnError(e -> failures.inc()).onErrorResume(e -> Mono.empty()); }

每次請求完成時，上述此輔助方法都會增加成功和失敗的次數。 而且，它記錄并吞下錯誤，因此單個錯誤或超時不會中斷整個導入過程。

private <T> Mono<T> countConcurrent(Mono<T> input) {return input.doOnSubscribe(s -> indexConcurrent.inc()).doOnTerminate(indexConcurrent::dec); }

上面的另一種方法是在發送新請求時增加indexConcurrent指標，并在結果或錯誤到達時將其遞減。 此指標不斷上升和下降，顯示進行中的請求數。

private <T> Mono<T> measure(Mono<T> input) {return Mono.fromCallable(indexTimer::time).flatMap(time ->input.doOnSuccess(x -> time.stop())); }

最終的助手方法是最復雜的。 它測量編制索引的總時間，即發送請求和接收響應之間的時間。 實際上，它非常通用，它只是計算訂閱任意Mono<T>到完成之間的總時間。 為什么看起來這么奇怪？ 好吧，基本的Timer API非常簡單

indexTimer.time(() -> someSlowCode())

它只需要一個lambda表達式并測量調用它花費了多長時間。 另外，您可以創建一個小的Timer.Context對象，該對象可以記住創建時間。 當您調用Context.stop()它將報告此度量：

final Timer.Context time = indexTimer.time(); someSlowCode(); time.stop();

使用異步流，要困難得多。 任務的開始（由預訂表示）和完成通常發生在代碼不同位置的線程邊界上。 我們可以做的是（懶惰地）創建一個新的Context對象（請參閱： fromCallable(indexTimer::time) ），并在包裝??的流完成時，完成Context （請參閱： input.doOnSuccess(x -> time.stop() ））。這是您構成所有這些方法的方式：

personGenerator.infinite().take(count).flatMap(doc -> countConcurrent(measure(indexDocSwallowErrors(doc))), concurrency);

就是這樣，但是用這么多低級的度量收集細節污染業務代碼似乎很奇怪。 讓我們用專門的組件包裝這些指標：

@RequiredArgsConstructor class EsMetrics {private final Timer indexTimer;private final Counter indexConcurrent;private final Counter successes;private final Counter failures;void success() {successes.inc();}void failure() {failures.inc();}void concurrentStart() {indexConcurrent.inc();}void concurrentStop() {indexConcurrent.dec();}Timer.Context startTimer() {return indexTimer.time();}}

現在，我們可以使用一些更高級的抽象：

class Indexer {private final EsMetrics esMetrics;private <T> Mono<T> countConcurrent(Mono<T> input) {return input.doOnSubscribe(s -> esMetrics.concurrentStart()).doOnTerminate(esMetrics::concurrentStop);}//...private Mono<IndexResponse> indexDocSwallowErrors(Doc doc) {return indexDoc(doc).doOnSuccess(response -> esMetrics.success()).doOnError(e -> log.error("Unable to index {}", doc, e)).doOnError(e -> esMetrics.failure()).onErrorResume(e -> Mono.empty());} }

在下一篇文章中，我們將學習如何更好地組合所有這些方法。 并避免一些樣板。

發布和可視化指標

僅僅收集指標是不夠的。 我們必須定期發布匯總指標，以便其他系統可以使用，處理和可視化它們。 一種這樣的工具是Graphite和Grafana 。 但是，在開始配置它們之前，讓我們首先將指標發布到控制臺。 我發現在對度量進行故障排除或開發時特別有用。

import com.codahale.metrics.MetricRegistry; import com.codahale.metrics.Slf4jReporter;@Bean Slf4jReporter slf4jReporter(MetricRegistry metricRegistry) {final Slf4jReporter slf4jReporter = Slf4jReporter.forRegistry(metricRegistry.build();slf4jReporter.start(1, TimeUnit.SECONDS);return slf4jReporter; }

這個簡單的代碼片段采用現有的MetricRegistry并注冊Slf4jReporter 。 每秒您將看到所有度量標準被打印到日志中（Logback等）：

type=COUNTER, name=es.concurrent, count=1 type=COUNTER, name=es.failures, count=0 type=COUNTER, name=es.successes, count=1653 type=TIMER, name=es.index, count=1653, min=1.104664, max=345.139385, mean=2.2166538118720576,stddev=11.208345077801448, median=1.455504, p75=1.660252, p95=2.7456, p98=5.625456, p99=9.69689, p999=85.062713,mean_rate=408.56403102372764, m1=0.0, m5=0.0, m15=0.0, rate_unit=events/second, duration_unit=milliseconds

但這僅僅是為了解決問題，為了將指標發布到外部Graphite實例，我們需要一個GraphiteReporter ：

import com.codahale.metrics.MetricRegistry; import com.codahale.metrics.graphite.Graphite; import com.codahale.metrics.graphite.GraphiteReporter;@Bean GraphiteReporter graphiteReporter(MetricRegistry metricRegistry) {final Graphite graphite = new Graphite(new InetSocketAddress("localhost", 2003));final GraphiteReporter reporter = GraphiteReporter.forRegistry(metricRegistry).prefixedWith("elastic-flux").convertRatesTo(TimeUnit.SECONDS).convertDurationsTo(TimeUnit.MILLISECONDS).build(graphite);reporter.start(1, TimeUnit.SECONDS);return reporter; }

在這里，我向localhost:2003報告，其中帶有Graphite + Grafana的Docker鏡像恰好在其中。 每秒將所有度量標準發送到該地址。 我們稍后可以在Grafana上可視化所有這些指標：

上圖顯示了索引時間分布（從第50個百分位數到第99.9個百分位數）。 使用此圖，您可以快速發現典型性能（P50）和（幾乎）最壞情況的性能（P99.9）。 對數標度是不尋常的，但是在這種情況下，我們可以看到上下百分位。 底部圖更加有趣。 它結合了三個指標：

• 成功執行索引操作的速率（每秒請求數）
• 操作失敗率（紅色條，堆疊在綠色條上）
• 當前并發級別（右軸）：進行中的請求數

此圖顯示了系統吞吐量（RPS），故障和并發性。 故障太多或并發級別異常高（許多操作正在等待響應）可能表明您的系統存在某些問題。 儀表板定義在GitHub存儲庫中可用。

在下一篇文章中，我們將學習如何從Dropwizard指標遷移到微米。 一個非常愉快的經歷！

翻譯自: https://www.javacodegeeks.com/2018/01/monitoring-measuring-reactive-application-dropwizard-metrics.html

總結

以上是生活随笔為你收集整理的使用Dropwizard度量标准监视和测量无功应用的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。