dubbo+zipkin呼叫鏈監控

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

收集器抽象

由於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();
    }
}

歡迎工作一到五年的Java工程師朋友們加入Java架構開發： 855835163 群內提供免費的Java架構學習資料（裡面有高可用、高併發、高效能及分散式、Jvm效能調優、Spring原始碼，MyBatis，Netty,Redis,Kafka,Mysql,Zookeeper,Tomcat,Docker,Dubbo,Nginx等多個知識點的架構資料）合理利用自己每一分每一秒的時間來學習提升自己，不要再用"沒有時間“來掩飾自己思想上的懶惰！趁年輕，使勁拼，給未來的自己一個交代！

dubbo+zipkin呼叫鏈監控

dubbo+zipkin呼叫鏈監控(二)

dubbo+zipkin呼叫鏈監控

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

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

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

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

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

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

dubbo+zipkin調用鏈監控

springcloud+zipkin實現鏈路監控

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

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

Dubbo呼叫過程監控

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

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

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

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

dubbo + zipkin 實現全鏈路追蹤

系統監控-分散式呼叫鏈Skywalking

CAT跨語言服務加拿大28平臺搭建鏈監控（七）消息分析器與報表

dubbo+zipkin呼叫鏈監控

相關推薦