2. 程式人生 > >dubbo+zipkin呼叫鏈監控(二)

dubbo+zipkin呼叫鏈監控(二)

阿新 • • 發佈：2018-12-14

去年的時候寫過dubbo+zipkin呼叫鏈監控,最近看到zipkin2配合brave實現起來會比我之前的實現要簡單很多，因為brave將很多互動的內容都封裝起來了，不需要自己去寫具體的實現，比如如何去構建span，如何去上報資料。

收集器抽象

由於zipkin支援http以及kafka兩種方式上報資料，所以在配置上需要做下抽象。

AbstractZipkinCollectorConfiguration

主要是針對下面兩種收集方式的一些配置上的定義，最核心的是Sender介面的定義，http與kafka是兩類完全不同的實現。

public abstract Sender getSender();

其次是協助性的建構函式，主要是配合構建收集器所需要的一些引數。

  • zipkinUrl

如果是http收集，那麼對應的是zipkin api域名，如果是kafka，對應的是kafka叢集的地址

  • topic

僅在收集方式為kafka是有效，http時傳空值即可。

public AbstractZipkinCollectorConfiguration(String serviceName,String zipkinUrl,String topic){
    this.zipkinUrl=zipkinUrl;
    this.serviceName=serviceName;
    this.topic=topic;
    this.tracing=this.tracing();
}

配置上報方式,這裡統一採用異常上傳，並且配置上報的超時時間。

protected AsyncReporter<Span> spanReporter() {
    return AsyncReporter
            .builder(getSender())
            .closeTimeout(500, TimeUnit.MILLISECONDS)
            .build(SpanBytesEncoder.JSON_V2);
}

下面這兩方法，是配合應用構建span使用的。

注意那個sampler()方法，預設是什麼也不做的意思，我們要想看到資料就需要配置成Sampler.ALWAYS_SAMPLE，這樣才能真正將資料上報到zipkin伺服器。

protected Tracing tracing() {
    this.tracing= Tracing
            .newBuilder()
            .localServiceName(this.serviceName)
            .sampler(Sampler.ALWAYS_SAMPLE)
            .spanReporter(spanReporter())
            .build();
    return this.tracing;
}

protected Tracing getTracing(){
    return this.tracing;
}

HttpZipkinCollectorConfiguration

主要是實現getSender方法，可以借用OkHttpSender這個物件來快速構建，api版本採用v2。

public class HttpZipkinCollectorConfiguration extends AbstractZipkinCollectorConfiguration {
    public HttpZipkinCollectorConfiguration(String serviceName,String zipkinUrl) {
        super(serviceName,zipkinUrl,null);
    }

    @Override
    public Sender getSender() {
        return OkHttpSender.create(super.getZipkinUrl()+"/api/v2/spans");
    }
}

OkHttpSender這個類需要引用這個包

<dependency>
    <groupId>io.zipkin.reporter2</groupId>
    <artifactId>zipkin-sender-okhttp3</artifactId>
    <version>${zipkin-reporter2.version}</version>
</dependency>

KafkaZipkinCollectorConfiguration

同樣也是實現getSender方法

public class KafkaZipkinCollectorConfiguration extends AbstractZipkinCollectorConfiguration {
    public KafkaZipkinCollectorConfiguration(String serviceName,String zipkinUrl,String topic) {
        super(serviceName,zipkinUrl,topic);
    }

    @Override
    public Sender getSender() {

        return KafkaSender
                .newBuilder()
                .bootstrapServers(super.getZipkinUrl())
                .topic(super.getTopic())
                .encoding(Encoding.JSON)
                .build();
    }
}

KafkaSender這個類需要引用這個包：

<dependency>
    <groupId>io.zipkin.reporter2</groupId>
    <artifactId>zipkin-sender-kafka11</artifactId>
    <version>${zipkin-reporter2.version}</version>
</dependency>

收集器工廠

由於上面建立了兩個收集器配置類，使用時只能是其中之一，所以實際執行的例項需要根據配置來動態生成。ZipkinCollectorConfigurationFactory就是負責生成收集器例項的。

private final AbstractZipkinCollectorConfiguration zipkinCollectorConfiguration;

@Autowired
public ZipkinCollectorConfigurationFactory(TraceConfig traceConfig){
    if(Objects.equal("kafka", traceConfig.getZipkinSendType())){
        zipkinCollectorConfiguration=new KafkaZipkinCollectorConfiguration(
                traceConfig.getApplicationName(),
                traceConfig.getZipkinUrl(),
                traceConfig.getZipkinKafkaTopic());
    }
    else {
        zipkinCollectorConfiguration = new HttpZipkinCollectorConfiguration(
                traceConfig.getApplicationName(),
                traceConfig.getZipkinUrl());
    }
}

通過構建函式將我們的配置類TraceConfig注入進來，然後根據傳送方式來構建例項。另外提供一個輔助函式：

public Tracing getTracing(){
    return this.zipkinCollectorConfiguration.getTracing();
}

過濾器

在dubbo的過濾器中實現資料上傳的功能邏輯相對簡單，一般都在invoke方法執行前記錄資料，然後方法執行完成後再次記錄資料。這個邏輯不變，有變化的是資料上報的實現，上一個版本是通過發http請求實現需要編碼，現在可以直接借用brave所提供的span來幫助我們完成，有兩重要的方法:

  • finish

方法原始碼如下，在完成的時候會填寫上完成的時間並上報資料，這一般應用於同步呼叫場景。

public void finish(TraceContext context, long finishTimestamp) {
    MutableSpan span = this.spanMap.remove(context);
    if(span != null && !this.noop.get()) {
        synchronized(span) {
            span.finish(Long.valueOf(finishTimestamp));
            this.reporter.report(span.toSpan());
        }
    }
}
  • flush 與上面finish方法的不同點在於，在報資料時沒有完成時間，這應該是適用於一些非同步呼叫但不關心結果的場景，比如dubbo所提供的oneway方式呼叫。
public void flush(TraceContext context) {
    MutableSpan span = this.spanMap.remove(context);
    if(span != null && !this.noop.get()) {
        synchronized(span) {
            span.finish((Long)null);
            this.reporter.report(span.toSpan());
        }
    }
}

消費者

做為消費方，有一個核心功能就是將traceId以及spanId傳遞到服務提供方，這裡還是通過dubbo提供的附加引數方式實現。

@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
    if(!RpcTraceContext.getTraceConfig().isEnabled()){
        return invoker.invoke(invocation);
    }

    ZipkinCollectorConfigurationFactory zipkinCollectorConfigurationFactory=
            SpringContextUtils.getApplicationContext().getBean(ZipkinCollectorConfigurationFactory.class);
    Tracer tracer= zipkinCollectorConfigurationFactory.getTracing().tracer();

    if(null==RpcTraceContext.getTraceId()){
        RpcTraceContext.start();
        RpcTraceContext.setTraceId(IdUtils.get());
        RpcTraceContext.setParentId(null);
        RpcTraceContext.setSpanId(IdUtils.get());
    }
    else {
        RpcTraceContext.setParentId(RpcTraceContext.getSpanId());
        RpcTraceContext.setSpanId(IdUtils.get());
    }
    TraceContext traceContext= TraceContext.newBuilder()
            .traceId(RpcTraceContext.getTraceId())
            .parentId(RpcTraceContext.getParentId())
            .spanId(RpcTraceContext.getSpanId())
            .sampled(true)
            .build();

    Span span=tracer.toSpan(traceContext).start();

    invocation.getAttachments().put(RpcTraceContext.TRACE_ID_KEY, String.valueOf(span.context().traceId()));
    invocation.getAttachments().put(RpcTraceContext.SPAN_ID_KEY, String.valueOf(span.context().spanId()));

    Result result = invoker.invoke(invocation);

    span.finish();

    return result;
}

提供者

@Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        if(!RpcTraceContext.getTraceConfig().isEnabled()){
            return invoker.invoke(invocation);
        }

        Map<String, String> attaches = invocation.getAttachments();
        if (!attaches.containsKey(RpcTraceContext.TRACE_ID_KEY)){
            return invoker.invoke(invocation);
        }

        Long traceId = Long.valueOf(attaches.get(RpcTraceContext.TRACE_ID_KEY));
        Long spanId = Long.valueOf(attaches.get(RpcTraceContext.SPAN_ID_KEY));

        attaches.remove(RpcTraceContext.TRACE_ID_KEY);
        attaches.remove(RpcTraceContext.SPAN_ID_KEY);
        RpcTraceContext.start();
        RpcTraceContext.setTraceId(traceId);
        RpcTraceContext.setParentId(spanId);
        RpcTraceContext.setSpanId(IdUtils.get());

        ZipkinCollectorConfigurationFactory zipkinCollectorConfigurationFactory=
                SpringContextUtils.getApplicationContext().getBean(ZipkinCollectorConfigurationFactory.class);
        Tracer tracer= zipkinCollectorConfigurationFactory.getTracing().tracer();

        TraceContext traceContext= TraceContext.newBuilder()
                .traceId(RpcTraceContext.getTraceId())
                .parentId(RpcTraceContext.getParentId())
                .spanId(RpcTraceContext.getSpanId())
                .sampled(true)
                .build();
        Span span = tracer.toSpan(traceContext).start();

        Result result = invoker.invoke(invocation);

        span.finish();

        return result;

    }

異常流程

上面無論是消費者的過濾器還是服務提供者的過濾器，均未考慮服務在呼叫invoker.invoke時出錯的場景，如果出錯，後面的span.finish方法將不會按預期執行，也就記錄不了資訊。所以需要針對此問題做優化：可以在finally塊中執行finish方法。

try {
    result = invoker.invoke(invocation);
}
finally {
    span.finish();
}

消費者在呼叫服務時，非同步呼叫問題

上面過濾器中呼叫span.finish都是基於同步模式，而由於dubbo除了同步呼叫外還提供了兩種呼叫方式

  • 非同步呼叫 通過callback機制的非同步

  • oneway

只發起請求並不等待結果的非同步呼叫，無callback一說

針對上面兩類非同步再加上同步呼叫，我們要想準確記錄服務真正的時間，需要在消費方的過濾器中做如下處理：

建立一個用於回撥的處理類，它的主要目的是為了在回撥成功時記錄時間，這裡無論是成功還是失敗。

private class AsyncSpanCallback implements ResponseCallback{

    private Span span;

    public AsyncSpanCallback(Span span){
        this.span=span;
    }

    @Override
    public void done(Object o) {
        span.finish();
    }

    @Override
    public void caught(Throwable throwable) {
        span.finish();
    }
}

再在呼叫invoke方法時,如果是oneway方式，則呼叫flush方法結果，如果是同步則直接呼叫finish方法，如果是非同步則在回撥時呼叫finish方法。


Result result = null;
boolean isOneway = RpcUtils.isOneway(invoker.getUrl(), invocation);
try {
    result = invoker.invoke(invocation);
}
finally {
    if(isOneway) {
        span.flush();
    }
    else if(!isAsync) {
        span.finish();
    }
}

待完善問題

過濾器中生成span的方式應該有更好的方法，還沒有對brave做過多研究，後續想辦法再優化下。另外我測試的場景是consumer呼叫provider，provider內部再呼叫provider2，我測試時發現第三步呼叫傳遞的parentId好像有點小問題，後續需要再確認下。

程式碼下載

相關推薦

dubbo+zipkin呼叫鏈監控(二)

去年的時候寫過dubbo+zipkin呼叫鏈監控,最近看到zipkin2配合brave實現起來會比我之前的實現要簡單很多，因為brave將很多互動的內容都封裝起來了，不需要自己去寫具體的實現，比如如何去構建span，如何去上報資料。 收集器抽象 由於zipkin支援http以及kafka兩種方式上

dubbo+zipkin呼叫鏈監控

收集器抽象 由於zipkin支援http以及kafka兩種方式上報資料，所以在配置上需要做下抽象。 AbstractZipkinCollectorConfiguration 主要是針對下面兩種收集方式的一些配置上的定義，最核心的是Sender介面的定

十二：Spring Cloud 之呼叫鏈監控-Sleuth

1. 簡介 Spring Cloud Sleuth implements a distributed tracing solution for Spring Cloud. 微服務的呼叫鏈監控是解

微服務-springCloud快速實踐2：服務監控、熔斷器監控及zipkin呼叫鏈

springCloud快速實踐2：服務監控、熔斷器監控及zipkin呼叫鏈 完整程式碼下載連結： https://github.com/2010yhh/springCloud-demos.git 環境 idea2018，jdk1.8， springboot版

幾種分散式呼叫鏈監控元件的實踐與比較（二）比較

引言：繼上篇《幾種分散式呼叫鏈監控元件的實踐與比較（一）實踐》後，本篇將會講下幾種APM選型的比

Spring Cloud Alibaba學習筆記（23） - 呼叫鏈監控工具Spring Cloud Sleuth + Zipkin

隨著業務發展，系統拆分導致系統呼叫鏈路愈發複雜一個前端請求可能最終需要呼叫很多次後端服務才能完成，當整個請求陷入效能瓶頸或不可用時，我們是無法得知該請求是由某個或某些後端服務引起的，這時就需要解決如何快讀定位服務故障點，以對症下藥。於是就有了分散式系統呼叫跟蹤的誕生。 Spring Cloud Sleuth

springmvc集成zipkin性能監控(二)

pro 添加 stat field 就是 com ren dispatch test6 上一遍是配置類，經過公司配置實踐，可能會給項目中原始的適配器產生沖突。 但是那樣會有一個好處，就是解耦合，不受本地開發環境和測試環境的影響，因為我們可以在部署的時候去簡單的去添加兩個配置

呼叫鏈系列二：解讀UAVStack中的呼叫鏈技術

在上週的推送中，我們介紹了呼叫鏈的模型設計與模型時序圖，本週將為大家繼續介紹呼叫鏈是如何在中介軟體增強技術的賦能之下實現服務端資訊收集以及服務間上下文傳遞的。 服務端資訊收集 服務端資訊收集流程如下圖所示，通過在應用容器（Tomcat等）啟動過程中植入切點，實現在應用邏輯執行之前和之後對請求進行劫持。

分散式呼叫鏈（二）---Control/Service層插樁埋點實際處理

為什麼要在Control/Service 層進行埋點 因為請求過來以後，先進入Control，在進入Service。 在Control：可以統計某個URL，具體的執行次數、時間、一段時間的流量統計。 在Service ：可以統計某個方法的用時。 當有了資料以後，可以使

十一：對微服務呼叫鏈監控的理解

微服務專欄地址 目錄 1. 簡介 微服務的呼叫鏈監控是解決微服務的複雜性帶來的一系列問題的強有效手段之一，從一下幾個方面來先理解微服務呼叫鏈監控相關： 簡介 什麼是呼叫鏈 為什麼要監控呼叫鏈 要監控哪些方面 呼叫鏈監控原理是什麼 5.1

微服務架構之「 呼叫鏈監控 」

「 呼叫鏈監控 」是在微服務興起後才有的一種新流行的監控模式。因為在我們傳統單體應用的專案中，不存在服務鏈/呼叫鏈的概念，所以也就根本沒有呼叫鏈監控的需求了。 當我們開始微服務架構之後，我們的很多服務變成分散式的了，並且我們對服務進行了拆分，拆分之後，使用者的一個請求進來，會依次經過不同的服務節點進行

dubbo+zipkin調用鏈監控

atop 動態生成 implement oid mov 都在 eid close void 圖片描述(最多50字)收集器抽象 由於zipkin支持http以及kafka兩種方式上報數據，所以在配置上需要做下抽象。 AbstractZipkinCollectorConfig

每天學點SpringCloud（十二）：Zipkin全鏈路監控

Zipkin是SpringCloud官方推薦的一款分散式鏈路監控的元件，使用它我們可以得知每一個請求所經過的節點以及耗時等資訊，並且它對程式碼無任何侵入，我們先來看一下Zipkin給我們提供的UI介面都是提供了哪些資訊。 zipkin首頁為我們提供了對於呼叫鏈路的搜尋查詢

zipkin+sleuth定時任務生成全域性呼叫鏈（十二）

問題背景 github提供的方式，是一個zipkin-dependencies-xxx.jar ，通過啟動這個jar包生成呼叫鏈，這個jar包啟動後，執行一次執行緒就會自動結束了，這在生產環境是不可忍受的。 準備 在github上下載最新的 release原始碼

springcloud+zipkin實現鏈路監控

轉載：http://ju.outofmemory.cn/entry/350377 Zipkin簡介 Zipkin是 Twitter 的一個 開源專案 ，它基於 Google Dapper實現。我們可以使用它來收集各個 伺服

zipkin使用hystrix無法串聯起呼叫鏈的解決方法

在使用zipkin的過程，引用hystrix導致呼叫鏈條break,原來hystrix預設容錯方法是執行緒隔離，而在這個過程中使用了執行緒池，而執行緒池中的traceId無法重新生成，導致呼叫鏈斷掉，因此重寫HystrixConcurrencyStrategy 類的wrapCallabl

Dubbo呼叫過程監控

MonitorFilter 主要對呼叫過程進行監控， public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException { if(invoker.ge

使用Zipkin和Brave 實現dubbo服務呼叫跟蹤

git專案地址：https://github.com/blacklau/http-dubbo-zipkin(點選開啟連結)，請下載使用。 本工程通過模擬訂單詳情的查詢，演示系統的呼叫鏈跟蹤，跟蹤資訊包括呼叫服務及請求引數。 涉及的各工程作用: louie-webap

原理分析dubbo分散式應用中使用zipkin做鏈路追蹤

zipkin是什麼 Zipkin是一款開源的分散式實時資料追蹤系統（Distributed Tracking System），基於 Google Dapper的論文設計而來，由 Twitter 公司開發貢獻。其主要功能是聚集來自各個異構系統的實時監控資料。分散式跟蹤系統還有其他比較成熟的實現，例如：Nave

使用zipkin+elasticsearech+brave對dubbo服務進行鏈路跟蹤

（一）我們先說下能為我們做什麼？ 1.zipkin為分散式鏈路呼叫監控系統，聚合各業務系統呼叫延遲資料，達到鏈路呼叫監控跟蹤； 2.zipkin系統讓開發者可通過一個Web前端輕鬆的收集和分析資料，例如使用者每次請求服務的處理時間等，可方便的監測系統中存在的瓶頸。 2.1服務