【zipkin/jaeger使用系列】zipkin与thrift结合进行接口监控

#参考资料

http://blog.mozhu.org/2017/11/11/zipkin/zipkin-2.html

进入正题

添加maven依赖

<dependency>
	<groupId>io.zipkin.brave</groupId>
	<artifactId>brave</artifactId>
	<version>5.0.0</version>
</dependency>
<dependency>
	<groupId>io.zipkin.reporter2</groupId>
	<artifactId>zipkin-reporter</artifactId>
	<version>2.7.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/io..reporter2/-sender-okhttp3 -->
<dependency>
	<groupId>io.zipkin.reporter2</groupId>
	<artifactId>zipkin-sender-okhttp3</artifactId>
	<version>2.7.0</version>
</dependency>

配置文件(用于初始化)

package com.jxdw.apigateway.config;

import brave.Tracer;
import brave.Tracing;
import brave.propagation.B3Propagation;
import brave.propagation.ExtraFieldPropagation;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;
import zipkin2.codec.SpanBytesEncoder;
import zipkin2.reporter.AsyncReporter;
import zipkin2.reporter.Sender;
import zipkin2.reporter.okhttp3.OkHttpSender;

import java.util.concurrent.TimeUnit;

@Component
public class TracerConfig implements InitializingBean {
    private static Tracer tracer;

    public static Tracer getTracer(){
        return tracer;
    }
    @Override
    public void afterPropertiesSet() throws Exception {
        Sender sender= OkHttpSender.create("http://192.168.172.6:9411/api/v2/spans");
        AsyncReporter asyncReporter=AsyncReporter.builder(sender).closeTimeout(500, TimeUnit.MILLISECONDS).build(SpanBytesEncoder.JSON_V2);
        Tracing tracing= Tracing.newBuilder().localServiceName("xiaogang_api").spanReporter(asyncReporter)
                .propagationFactory(ExtraFieldPropagation.newFactory(B3Propagation.FACTORY,"user-name")).build();
        tracer=tracing.tracer();
    }
}

嵌入到使用的rpc框架入口（多服务共有一个端口）

package com.jxdw.apigateway.thrift;

import brave.Span;
import com.jxdw.apigateway.config.TracerConfig;

import org.apache.thrift.TException;
import org.apache.thrift.TProcessor;

import org.apache.thrift.protocol.TMessage;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.protocol.TProtocolDecorator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;

public class MyTMutiplexedProcessor implements TProcessor {
    private static final Logger logger = LoggerFactory.getLogger(ThriftPoolServerStartThread.class);
    private final Map<String, TProcessor> SERVICE_PROCESSOR_MAP = new HashMap();


    public void registerProcessor(String serviceName, TProcessor processor) {
        this.SERVICE_PROCESSOR_MAP.put(serviceName, processor);
    }
    @Override
    public boolean process(TProtocol inProtocol, TProtocol outProtocol1) throws TException {
        TMessage message = inProtocol.readMessageBegin();
        logger.info("service name {}",message.name);
        if (message.type != 1 && message.type != 4) {
            throw new TException("This should not have happened!?");
        } else {
            int index = message.name.indexOf(":");
            if (index < 0) {
                throw new TException("Service name not found in message name: " + message.name + ".  Did you " + "forget to use a TMultiplexProtocol in your client?");
            } else {
                String serviceName = message.name.substring(0, index);

                TProcessor actualProcessor = (TProcessor)this.SERVICE_PROCESSOR_MAP.get(serviceName);
                if (actualProcessor == null) {
                    throw new TException("Service name not found: " + serviceName + ".  Did you forget " + "to call registerProcessor()?");
                } else {
                    logger.info("rpc start");
					TMessage standardMessage = new TMessage(message.name.substring(serviceName.length() + ":".length()), message.type, message.seqid);
                    Span span=TracerConfig.getTracer().newTrace().name(message.name.substring(serviceName.length() + ":".length())).start();
                     boolean result=actualProcessor.process(new MyTMutiplexedProcessor.StoredMessageProtocol(inProtocol, standardMessage), outProtocol1);
                    span.finish();
					logger.info("rpc end");
                    return result;
                }
            }
        }
    }

    private static class StoredMessageProtocol extends TProtocolDecorator {
        TMessage messageBegin;

        public StoredMessageProtocol(TProtocol protocol, TMessage messageBegin) {
            super(protocol);
            this.messageBegin = messageBegin;
        }

        public TMessage readMessageBegin() throws TException {
            return this.messageBegin;
        }
    }
}

效果

题外话

以前的技术选型做的有点草率，thrift这个框架可定制化的地方不是很多。