採用brave接入zipkin實現自定義rpc跟蹤

本文假設你已經閱讀過Dapper文章：http://bigbully.github.io/Dapper-translation/，對分佈式服務跟蹤有了初步瞭解。另外一篇zipkin的介紹文章（基礎篇）能夠參考：http://www.tangrui.net/implement-distributed-tracking-system-with-zipkin-and-brave/?utm_source=tuicool&utm_medium=referral

本文旨在介紹最新的Brave版本的使用，如何模擬出Dapper的調用鏈效果。

實現效果

定義有兩個模塊之間的調用，appserver服務調用datacenter服務。

其中由外部網絡調用appserver的group_data接口，而後group_data再分別調用datacenter的get_radio_list、get_user_list、get_program_list三個接口。

代碼

如今，爲了模擬出這個調用層級，給你們理解brave的使用。給出了模擬的代碼例子：

import java.util.ArrayList;
import java.util.Collection;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

import com.github.kristofa.brave.Brave;
import com.github.kristofa.brave.ClientRequestAdapter;
import com.github.kristofa.brave.ClientRequestInterceptor;
import com.github.kristofa.brave.ClientResponseAdapter;
import com.github.kristofa.brave.ClientResponseInterceptor;
import com.github.kristofa.brave.EmptySpanCollectorMetricsHandler;
import com.github.kristofa.brave.KeyValueAnnotation;
import com.github.kristofa.brave.ServerRequestAdapter;
import com.github.kristofa.brave.ServerRequestInterceptor;
import com.github.kristofa.brave.ServerResponseAdapter;
import com.github.kristofa.brave.ServerResponseInterceptor;
import com.github.kristofa.brave.SpanId;
import com.github.kristofa.brave.TraceData;
import com.github.kristofa.brave.http.HttpSpanCollector;
import com.twitter.zipkin.gen.Endpoint;

public class BraveTest {
	
	private static HttpSpanCollector collector = null;
	private static Brave brave = null;
	private static Brave brave2 = null;
	
	private static void braveInit(){
		collector = HttpSpanCollector.create("http://localhost:9411/", new EmptySpanCollectorMetricsHandler());

		brave = new Brave.Builder("appserver").spanCollector(collector).build();
		brave2 = new Brave.Builder("datacenter").spanCollector(collector).build();
	}
	
	static class Task {
		String name;
		SpanId spanId;
		public Task(String name, SpanId spanId) {
			super();
			this.name = name;
			this.spanId = spanId;
		}
	}

	public static void main(String[] args) throws Exception {
		braveInit();
		
		final BlockingQueue<Task> queue = new ArrayBlockingQueue<Task>(10);
		Thread thread = new Thread(){
			public void run() {
				while (true) {
					try {
						Task task = queue.take();
						dcHandle(task.name, task.spanId);
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			}
		};
		thread.start();
		
		{
			ServerRequestInterceptor serverRequestInterceptor = brave.serverRequestInterceptor();
			ServerResponseInterceptor serverResponseInterceptor = brave.serverResponseInterceptor();
			ClientRequestInterceptor clientRequestInterceptor = brave.clientRequestInterceptor();
			ClientResponseInterceptor clientResponseInterceptor = brave.clientResponseInterceptor();
			
			serverRequestInterceptor.handle(new ServerRequestAdapterImpl("group_data"));
			
			ClientRequestAdapterImpl clientRequestAdapterImpl = new ClientRequestAdapterImpl("get_radio_list");
			clientRequestInterceptor.handle(clientRequestAdapterImpl);
			queue.offer(new Task("get_radio_list", clientRequestAdapterImpl.getSpanId()));
			Thread.sleep(10);
			clientResponseInterceptor.handle(new ClientResponseAdapterImpl());
			
			clientRequestAdapterImpl = new ClientRequestAdapterImpl("get_user_list");
			clientRequestInterceptor.handle(clientRequestAdapterImpl);
			queue.offer(new Task("get_user_list", clientRequestAdapterImpl.getSpanId()));
			Thread.sleep(10);
			clientResponseInterceptor.handle(new ClientResponseAdapterImpl());
			
			clientRequestAdapterImpl = new ClientRequestAdapterImpl("get_program_list");
			clientRequestInterceptor.handle(clientRequestAdapterImpl);
			queue.offer(new Task("get_program_list", clientRequestAdapterImpl.getSpanId()));
			Thread.sleep(10);
			clientResponseInterceptor.handle(new ClientResponseAdapterImpl());
			
			serverResponseInterceptor.handle(new ServerResponseAdapterImpl());
		}
		Thread.sleep(3000);
	}
	
	public static void dcHandle(String spanName, SpanId spanId){
		ServerRequestInterceptor serverRequestInterceptor = brave2.serverRequestInterceptor();
		ServerResponseInterceptor serverResponseInterceptor = brave2.serverResponseInterceptor();
		serverRequestInterceptor.handle(new ServerRequestAdapterImpl(spanName, spanId));
		serverResponseInterceptor.handle(new ServerResponseAdapterImpl());
	}
	
	
	static class ServerRequestAdapterImpl implements ServerRequestAdapter {
		
		Random randomGenerator = new Random();
		SpanId spanId;
		String spanName;
		
		ServerRequestAdapterImpl(String spanName){
			this.spanName = spanName;
			long startId = randomGenerator.nextLong();
			SpanId spanId = SpanId.builder().spanId(startId).traceId(startId).parentId(startId).build();
			this.spanId = spanId;
		}
		
		ServerRequestAdapterImpl(String spanName, SpanId spanId){
			this.spanName = spanName;
			this.spanId = spanId;
		}

		@Override
		public TraceData getTraceData() {
			if (this.spanId != null) {
				return TraceData.builder().spanId(this.spanId).build();
			}
			long startId = randomGenerator.nextLong();
			SpanId spanId = SpanId.builder().spanId(startId).traceId(startId).parentId(startId).build();
			return TraceData.builder().spanId(spanId).build();
		}

		@Override
		public String getSpanName() {
			return spanName;
		}

		@Override
		public Collection<KeyValueAnnotation> requestAnnotations() {
			Collection<KeyValueAnnotation> collection = new ArrayList<KeyValueAnnotation>();
			KeyValueAnnotation kv = KeyValueAnnotation.create("radioid", "165646485468486364");
			collection.add(kv);
			return collection;
		}
		
	}
	
	static class ServerResponseAdapterImpl implements ServerResponseAdapter {

		@Override
		public Collection<KeyValueAnnotation> responseAnnotations() {
			Collection<KeyValueAnnotation> collection = new ArrayList<KeyValueAnnotation>();
			KeyValueAnnotation kv = KeyValueAnnotation.create("radioid", "165646485468486364");
			collection.add(kv);
			return collection;
		}
		
	}
	
	static class ClientRequestAdapterImpl implements ClientRequestAdapter {
		
		String spanName;
		SpanId spanId;
		
		ClientRequestAdapterImpl(String spanName){
			this.spanName = spanName;
		}
		
		public SpanId getSpanId() {
			return spanId;
		}

		@Override
		public String getSpanName() {
			return this.spanName;
		}

		@Override
		public void addSpanIdToRequest(SpanId spanId) {
			//記錄傳輸到遠程服務
			System.out.println(spanId);
			if (spanId != null) {
				this.spanId = spanId;
				System.out.println(String.format("trace_id=%s, parent_id=%s, span_id=%s", spanId.traceId, spanId.parentId, spanId.spanId));
			}
		}

		@Override
		public Collection<KeyValueAnnotation> requestAnnotations() {
			Collection<KeyValueAnnotation> collection = new ArrayList<KeyValueAnnotation>();
			KeyValueAnnotation kv = KeyValueAnnotation.create("radioid", "165646485468486364");
			collection.add(kv);
			return collection;
		}

		@Override
		public Endpoint serverAddress() {
			return null;
		}
		
	}

	static class ClientResponseAdapterImpl implements ClientResponseAdapter {

		@Override
		public Collection<KeyValueAnnotation> responseAnnotations() {
			Collection<KeyValueAnnotation> collection = new ArrayList<KeyValueAnnotation>();
			KeyValueAnnotation kv = KeyValueAnnotation.create("radioname", "火星人1");
			collection.add(kv);
			return collection;
		}
		
	}
}

maven的依賴包：

<dependency>
		<groupId>io.zipkin.brave</groupId>
		<artifactId>brave-spancollector-http</artifactId>
		<version>3.9.0</version>
	</dependency>

結果

把下載好的zipkin-server.jar包運行起來：java -jar zipkin-server.jar

而後運行上面測試代碼後，用瀏覽器打開web UI：http://localhost:9411/

點擊上面紅色方框的記錄

能夠看到每一個請求消耗的時間和調用順序，點擊第一個datacenter的調用記錄，顯示以下：

上面的彈窗框展現了一個調用從Client Send->Server Receive->Server Send->Client Receive的完整耗時。(Relative Time相對時間)