<parent>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-parent</artifactId>

<version>1.3.6.RELEASE</version>

</parent>

<!-- https://mvnrepository.com/artifact/io.zipkin.brave/brave-core -->

<dependencies>

<dependency>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-web</artifactId>

</dependency>

<dependency>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-aop</artifactId>

</dependency>

<dependency>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-actuator</artifactId>

</dependency>

<dependency>

<groupId>io.zipkin.brave</groupId>

<artifactId>brave-core</artifactId>

<version>3.9.0</version>

</dependency>

<!-- https://mvnrepository.com/artifact/io.zipkin.brave/brave-http -->

<dependency>

<groupId>io.zipkin.brave</groupId>

<artifactId>brave-http</artifactId>

<version>3.9.0</version>

</dependency>

<dependency>

<groupId>io.zipkin.brave</groupId>

<artifactId>brave-spancollector-http</artifactId>

<version>3.9.0</version>

</dependency>

<dependency>

<groupId>io.zipkin.brave</groupId>

<artifactId>brave-web-servlet-filter</artifactId>

<version>3.9.0</version>

</dependency>

<dependency>

<groupId>io.zipkin.brave</groupId>

<artifactId>brave-okhttp</artifactId>

<version>3.9.0</version>

</dependency>

<!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-api -->

<dependency>

<groupId>org.slf4j</groupId>

<artifactId>slf4j-api</artifactId>

<version>1.7.13</version>

</dependency>

<dependency>

<groupId>org.apache.httpcomponents</groupId>

<artifactId>httpclient</artifactId>

<version>4.5.1</version>

</dependency>

</dependencies>

package com.lkl.zipkin;

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

/**

*

* Created by liaokailin on 16/7/27.

*/

@SpringBootApplication

public class Application {

public static void main(String[] args) {

SpringApplication app = new SpringApplication(Application.class);

app.run(args);

}

}

RestController

@RequestMapping("/")

public class HomeController {

@Autowired

private OkHttpClient client;

private Random random = new Random();

@RequestMapping("start")

public String start() throws InterruptedException, IOException {

int sleep= random.nextInt(100);

TimeUnit.MILLISECONDS.sleep(sleep);

Request request = new Request.Builder().url("http://localhost:9090/foo").get().build();

Response response = client.newCall(request).execute();

return " [service1 sleep " + sleep+" ms]" + response.body().toString();

}

package com.lkl.zipkin.config;

import com.github.kristofa.brave.Brave;

import com.github.kristofa.brave.EmptySpanCollectorMetricsHandler;

import com.github.kristofa.brave.SpanCollector;

import com.github.kristofa.brave.http.DefaultSpanNameProvider;

import com.github.kristofa.brave.http.HttpSpanCollector;

import com.github.kristofa.brave.okhttp.BraveOkHttpRequestResponseInterceptor;

import com.github.kristofa.brave.servlet.BraveServletFilter;

import okhttp3.OkHttpClient;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

/**

* Created by liaokailin on 16/7/27.

*/

@Configuration

public class ZipkinConfig {

@Autowired

private ZipkinProperties properties;

@Bean

public SpanCollector spanCollector() {

HttpSpanCollector.Config config = HttpSpanCollector.Config.builder().connectTimeout(properties.getConnectTimeout()).readTimeout(properties.getReadTimeout())

.compressionEnabled(properties.isCompressionEnabled()).flushInterval(properties.getFlushInterval()).build();

return HttpSpanCollector.create(properties.getUrl(), config, new EmptySpanCollectorMetricsHandler());

}

@Bean

public Brave brave(SpanCollector spanCollector){

Brave.Builder builder = new Brave.Builder(properties.getServiceName()); //指定state

builder.spanCollector(spanCollector);

builder.traceSampler(Sampler.ALWAYS_SAMPLE);

Brave brave = builder.build();

return brave;

}

@Bean

public BraveServletFilter braveServletFilter(Brave brave){

BraveServletFilter filter = new BraveServletFilter(brave.serverRequestInterceptor(),brave.serverResponseInterceptor(),new DefaultSpanNameProvider());

return filter;

}

@Bean

public OkHttpClient okHttpClient(Brave brave){

OkHttpClient client = new OkHttpClient.Builder()

.addInterceptor(new BraveOkHttpRequestResponseInterceptor(brave.clientRequestInterceptor(), brave.clientResponseInterceptor(), new DefaultSpanNameProvider()))

.build();

return client;

}

}

package com.lkl.zipkin.config;

import org.springframework.boot.context.properties.ConfigurationProperties;

import org.springframework.context.annotation.Configuration;

/**

* Created by liaokailin on 16/7/28.

*/

@Configuration

@ConfigurationProperties(prefix = "com.zipkin")

public class ZipkinProperties {

private String serviceName;

private String url;

private int connectTimeout;

private int readTimeout;

private int flushInterval;

private boolean compressionEnabled;

public String getUrl() {

return url;

}

public void setUrl(String url) {

this.url = url;

}

public int getConnectTimeout() {

return connectTimeout;

}

public void setConnectTimeout(int connectTimeout) {

this.connectTimeout = connectTimeout;

}

public int getReadTimeout() {

return readTimeout;

}

public void setReadTimeout(int readTimeout) {

this.readTimeout = readTimeout;

}

public int getFlushInterval() {

return flushInterval;

}

public void setFlushInterval(int flushInterval) {

this.flushInterval = flushInterval;

}

public boolean isCompressionEnabled() {

return compressionEnabled;

}

public void setCompressionEnabled(boolean compressionEnabled) {

this.compressionEnabled = compressionEnabled;

}

public String getServiceName() {

return serviceName;

}

public void setServiceName(String serviceName) {

this.serviceName = serviceName;

}

}

com.zipkin.serviceName=service1

com.zipkin.url=http://110.173.14.57:9411

com.zipkin.connectTimeout=6000

com.zipkin.readTimeout=6000

com.zipkin.flushInterval=1

com.zipkin.compressionEnabled=true

server.port=8080

com.zipkin.serviceName=service2

com.zipkin.url=http://110.173.14.57:9411

com.zipkin.connectTimeout=6000

com.zipkin.readTimeout=6000

com.zipkin.flushInterval=1

com.zipkin.compressionEnabled=true

server.port=9090

@RequestMapping("foo")

public String foo() throws InterruptedException, IOException {

Random random = new Random();

int sleep= random.nextInt(100);

TimeUnit.MILLISECONDS.sleep(sleep);

Request request = new Request.Builder().url("http://localhost:9091/bar").get().build(); //service3

Response response = client.newCall(request).execute();

String result = response.body().string();

request = new Request.Builder().url("http://localhost:9092/tar").get().build(); //service4

response = client.newCall(request).execute();

result += response.body().string();

return " [service2 sleep " + sleep+" ms]" + result;

}

@RequestMapping("bar")

public String bar() throws InterruptedException, IOException { //service3 method

Random random = new Random();

int sleep= random.nextInt(100);

TimeUnit.MILLISECONDS.sleep(sleep);

return " [service3 sleep " + sleep+" ms]";

}

@RequestMapping("tar")

public String tar() throws InterruptedException, IOException { //service4 method

Random random = new Random();

int sleep= random.nextInt(1000);

TimeUnit.MILLISECONDS.sleep(sleep);

return " [service4 sleep " + sleep+" ms]";

}

nohup java -jar server4.jar &

nohup java -jar server3.jar &

nohup java -jar server2.jar &

nohup java -jar server1.jar &

@Bean

public SpanCollector spanCollector() {

HttpSpanCollector.Config config = HttpSpanCollector.Config.builder().connectTimeout(properties.getConnectTimeout()).readTimeout(properties.getReadTimeout())

.compressionEnabled(properties.isCompressionEnabled()).flushInterval(properties.getFlushInterval()).build();

return HttpSpanCollector.create(properties.getUrl(), config, new EmptySpanCollectorMetricsHandler());

}

@Override

protected void sendSpans(byte[] json) throws IOException {

// intentionally not closing the connection, so as to use keep-alives

HttpURLConnection connection = (HttpURLConnection) new URL(url).openConnection();

connection.setConnectTimeout(config.connectTimeout());

connection.setReadTimeout(config.readTimeout());

connection.setRequestMethod("POST");

connection.addRequestProperty("Content-Type", "application/json");

if (config.compressionEnabled()) {

connection.addRequestProperty("Content-Encoding", "gzip");

ByteArrayOutputStream gzipped = new ByteArrayOutputStream();

try (GZIPOutputStream compressor = new GZIPOutputStream(gzipped)) {

compressor.write(json);

}

json = gzipped.toByteArray();

}

connection.setDoOutput(true);

connection.setFixedLengthStreamingMode(json.length);

connection.getOutputStream().write(json);

try (InputStream in = connection.getInputStream()) {

while (in.read() != -1) ; // skip

} catch (IOException e) {

try (InputStream err = connection.getErrorStream()) {

if (err != null) { // possible, if the connection was dropped

while (err.read() != -1) ; // skip

}

}

throw e;

}

}

}

@Bean

public Brave brave(SpanCollector spanCollector){

Brave.Builder builder = new Brave.Builder(properties.getServiceName()); //指定state

builder.spanCollector(spanCollector);

builder.traceSampler(Sampler.ALWAYS_SAMPLE);

Brave brave = builder.build();

return brave;

}

public Brave build() {

return new Brave(this);

}

private Brave(Builder builder) {

serverTracer = ServerTracer.builder()

.randomGenerator(builder.random)

.spanCollector(builder.spanCollector)

.state(builder.state)

.traceSampler(builder.sampler).build();

clientTracer = ClientTracer.builder()

.randomGenerator(builder.random)

.spanCollector(builder.spanCollector)

.state(builder.state)

.traceSampler(builder.sampler).build();

localTracer = LocalTracer.builder()

.randomGenerator(builder.random)

.spanCollector(builder.spanCollector)

.spanAndEndpoint(SpanAndEndpoint.LocalSpanAndEndpoint.create(builder.state))

.traceSampler(builder.sampler).build();

serverRequestInterceptor = new ServerRequestInterceptor(serverTracer);

serverResponseInterceptor = new ServerResponseInterceptor(serverTracer);

clientRequestInterceptor = new ClientRequestInterceptor(clientTracer);

clientResponseInterceptor = new ClientResponseInterceptor(clientTracer);

serverSpanAnnotationSubmitter = AnnotationSubmitter.create(SpanAndEndpoint.ServerSpanAndEndpoint.create(builder.state));

serverSpanThreadBinder = new ServerSpanThreadBinder(builder.state);

clientSpanThreadBinder = new ClientSpanThreadBinder(builder.state);

}

@Override

public void doFilter(ServletRequest request, ServletResponse response, FilterChain filterChain) throws IOException, ServletException {

String alreadyFilteredAttributeName = getAlreadyFilteredAttributeName();

boolean hasAlreadyFilteredAttribute = request.getAttribute(alreadyFilteredAttributeName) != null;

if (hasAlreadyFilteredAttribute) {

// Proceed without invoking this filter...

filterChain.doFilter(request, response);

} else {

final StatusExposingServletResponse statusExposingServletResponse = new StatusExposingServletResponse((HttpServletResponse) response);

requestInterceptor.handle(new HttpServerRequestAdapter(new ServletHttpServerRequest((HttpServletRequest) request), spanNameProvider));

try {

filterChain.doFilter(request, statusExposingServletResponse);

} finally {

responseInterceptor.handle(new HttpServerResponseAdapter(new HttpResponse() {

@Override

public int getHttpStatusCode() {

return statusExposingServletResponse.getStatus();

}

}));

}

}

}

public void handle(ServerRequestAdapter adapter) {

serverTracer.clearCurrentSpan();

final TraceData traceData = adapter.getTraceData();

Boolean sample = traceData.getSample();

if (sample != null && Boolean.FALSE.equals(sample)) {

serverTracer.setStateNoTracing();

LOGGER.fine("Received indication that we should NOT trace.");

} else {

if (traceData.getSpanId() != null) {

LOGGER.fine("Received span information as part of request.");

SpanId spanId = traceData.getSpanId();

serverTracer.setStateCurrentTrace(spanId.traceId, spanId.spanId,

spanId.nullableParentId(), adapter.getSpanName());

} else {

LOGGER.fine("Received no span state.");

serverTracer.setStateUnknown(adapter.getSpanName());

}

serverTracer.setServerReceived();

for(KeyValueAnnotation annotation : adapter.requestAnnotations())

{

serverTracer.submitBinaryAnnotation(annotation.getKey(), annotation.getValue());

}

}

}

@Override

public void setCurrentServerSpan(final ServerSpan span) {

if (span == null) {

currentServerSpan.remove();

} else {

currentServerSpan.set(span);

}

}

@Override

public TraceData getTraceData() {

final String sampled = serverRequest.getHttpHeaderValue(BraveHttpHeaders.Sampled.getName());

if (sampled != null) {

if (sampled.equals("0") || sampled.toLowerCase().equals("false")) {

return TraceData.builder().sample(false).build();

} else {

final String parentSpanId = serverRequest.getHttpHeaderValue(BraveHttpHeaders.ParentSpanId.getName());

final String traceId = serverRequest.getHttpHeaderValue(BraveHttpHeaders.TraceId.getName());

final String spanId = serverRequest.getHttpHeaderValue(BraveHttpHeaders.SpanId.getName());

if (traceId != null && spanId != null) {

SpanId span = getSpanId(traceId, spanId, parentSpanId);

return TraceData.builder().sample(true).spanId(span).build();

}

}

}

return TraceData.builder().build();

}

private SpanId getSpanId(String traceId, String spanId, String parentSpanId) {

return SpanId.builder()

.traceId(convertToLong(traceId))

.spanId(convertToLong(spanId))

.parentId(parentSpanId == null ? null : convertToLong(parentSpanId)).build();

}

public Builder parentId(@Nullable Long parentId) {

if (parentId == null) {

this.flags |= FLAG_IS_ROOT;

} else {

this.flags &= ~FLAG_IS_ROOT;

}

this.parentId = parentId;

return this;

}

serverTracer.setStateCurrentTrace(spanId.traceId, spanId.spanId,

spanId.nullableParentId(), adapter.getSpanName());

public void setStateCurrentTrace(long traceId, long spanId, @Nullable Long parentSpanId, @Nullable String name) {

checkNotBlank(name, "Null or blank span name");

spanAndEndpoint().state().setCurrentServerSpan(

ServerSpan.create(traceId, spanId, parentSpanId, name));

}

static ServerSpan create(long traceId, long spanId, @Nullable Long parentSpanId, String name) {

Span span = new Span();

span.setTrace_id(traceId);

span.setId(spanId);

if (parentSpanId != null) {

span.setParent_id(parentSpanId);

}

span.setName(name);

return create(span, true);

}

public void setServerReceived() {

submitStartAnnotation(zipkinCoreConstants.SERVER_RECV);

}

void submitStartAnnotation(String annotationName) {

Span span = spanAndEndpoint().span();

if (span != null) {

Annotation annotation = Annotation.create(

currentTimeMicroseconds(),

annotationName,

spanAndEndpoint().endpoint()

);

synchronized (span) {

span.setTimestamp(annotation.timestamp);

span.addToAnnotations(annotation);

}

}

}

for(KeyValueAnnotation annotation : adapter.requestAnnotations())

{

serverTracer.submitBinaryAnnotation(annotation.getKey(), annotation.getValue());

}

@Override

public Collection<KeyValueAnnotation> requestAnnotations() {

KeyValueAnnotation uriAnnotation = KeyValueAnnotation.create(

TraceKeys.HTTP_URL, serverRequest.getUri().toString());

return Collections.singleton(uriAnnotation);

}

responseInterceptor.handle(new HttpServerResponseAdapter(new HttpResponse() {

@Override //獲取http狀態(tài)碼

public int getHttpStatusCode() {

return statusExposingServletResponse.getStatus();

}

}));

public void handle(ServerResponseAdapter adapter) {

// We can submit this in any case. When server state is not set or

// we should not trace this request nothing will happen.

LOGGER.fine("Sending server send.");

try {

for(KeyValueAnnotation annotation : adapter.responseAnnotations())

{

serverTracer.submitBinaryAnnotation(annotation.getKey(), annotation.getValue());

}

serverTracer.setServerSend();

} finally {

serverTracer.clearCurrentSpan();

}

}

public void setServerSend() {

if (submitEndAnnotation(zipkinCoreConstants.SERVER_SEND, spanCollector())) {

spanAndEndpoint().state().setCurrentServerSpan(null);

}

}

boolean submitEndAnnotation(String annotationName, SpanCollector spanCollector) {

Span span = spanAndEndpoint().span();

if (span == null) {

return false;

}

Annotation annotation = Annotation.create(

currentTimeMicroseconds(),

annotationName,

spanAndEndpoint().endpoint()

);

span.addToAnnotations(annotation);

if (span.getTimestamp() != null) {

span.setDuration(annotation.timestamp - span.getTimestamp());

}

spanCollector.collect(span);

return true;

}

@Override

public void collect(Span span) {

metrics.incrementAcceptedSpans(1);

if (!pending.offer(span)) {

metrics.incrementDroppedSpans(1);

}

}

static final class Flusher implements Runnable {

final Flushable flushable;

final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

Flusher(Flushable flushable, int flushInterval) {

this.flushable = flushable;

this.scheduler.scheduleWithFixedDelay(this, 0, flushInterval, SECONDS);

}

@Override

public void run() {

try {

flushable.flush();

} catch (IOException ignored) {

}

}

}

@Override

public void flush() {

if (pending.isEmpty()) return;

List<Span> drained = new ArrayList<Span>(pending.size());

pending.drainTo(drained);

if (drained.isEmpty()) return;

int spanCount = drained.size();

try {

reportSpans(drained);

} catch (IOException e) {

metrics.incrementDroppedSpans(spanCount);

} catch (RuntimeException e) {

metrics.incrementDroppedSpans(spanCount);

}

}

@Override

protected void reportSpans(List<Span> drained) throws IOException {

byte[] encoded = codec.writeSpans(drained);

sendSpans(encoded);

}