前面几篇博文中，都是使用OkHttpSender来上报Trace信息给Zipkin，这在生产环境中，当业务量比较大的时候，可能会成为一个性能瓶颈，这一篇博文我们来使用KafkaSender将Trace信息先写入到Kafka中，然后Zipkin使用KafkaCollector从Kafka中收集Span信息。
在Brave配置中需要将Sender设置为KafkaSender，而zipkin的collector组件配置为KafkaCollector

相关代码在Chapter8/zipkin-kafka中
pom.xml中添加依赖

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

TracingConfiguration中，我们修改Sender为KafkaSender，指定Kafka的地址，以及topic

@Bean
Sender sender() {
    return KafkaSender.newBuilder().bootstrapServers("localhost:9091,localhost:9092,localhost:9093").topic("zipkin").encoding(Encoding.JSON).build();
}

我们先启动zookeeper（默认端口号为2181），再依次启动一个本地的3个broker的kafka集群（端口号分别为9091、9092、9093），最后启动一个KafkaManager（默认端口号9000），KafkaManager是Kafka的UI管理工具
关于如何搭建本地Kafka伪集群，请自行上网搜索教程，本文使用的Kafka版本为0.10.0.0。

kafka启动完毕后，我们创建名为zipkin的topic，因为我们有3个broker，我这里设置replication-factor=3

bin/windows/kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic zipkin

打开KafkaManager界面

http://localhost:9000/clusters/localhost/topics/zipkin

可以看到topic zipkin中暂时没有消息。

我们使用如下命令启动zipkin，带上Kafka的Zookeeper地址参数，这样zipkin就会从kafka中消费我们上报的trace信息。

java -jar zipkin-server-2.2.1-exec.jar --KAFKA_ZOOKEEPER=localhost:2181

然后分别运行，主意我们这里将backend的端口改为9001，目的是为了避免和KafkaManager端口号冲突。

mvn spring-boot:run -Drun.jvmArguments="-Dserver.port=9001 -Dzipkin.service=backend"

mvn spring-boot:run -Drun.jvmArguments="-Dserver.port=8081 -Dzipkin.service=frontend"

浏览器访问 http://localhost:8081/ 会显示当前时间

我们再次刷新KafkaManager界面

http://localhost:9000/clusters/localhost/topics/zipkin

可以看到topic zipkin中有两条消息。

为了看到这两条消息的具体内容，我们可以在kafka安装目录使用如下命令

bin/windows/kafka-console-consumer.bat --zookeeper localhost:2181 --topic zipkin --from-beginning

在控制台会打印出最近的两条消息

[{"traceId":"802bd09f480b5faa","parentId":"802bd09f480b5faa","id":"bb3c70909ea3ee3c","kind":"SERVER","name":"get","timestamp":1510891296426607,"duration":10681,"localEndpoint":{"serviceName":"backend","ipv4":"10.200.170.137"},"remoteEndpoint":{"ipv4":"127.0.0.1","port":64421},"tags":{"http.path":"/api"},"shared":true}]
[{"traceId":"802bd09f480b5faa","parentId":"802bd09f480b5faa","id":"bb3c70909ea3ee3c","kind":"CLIENT","name":"get","timestamp":1510891296399882,"duration":27542,"localEndpoint":{"serviceName":"frontend","ipv4":"10.200.170.137"},"tags":{"http.path":"/api"}},{"traceId":"802bd09f480b5faa","id":"802bd09f480b5faa","kind":"SERVER","name":"get","timestamp":1510891296393252,"duration":39514,"localEndpoint":{"serviceName":"frontend","ipv4":"10.200.170.137"},"remoteEndpoint":{"ipv6":"::1","port":64420},"tags":{"http.path":"/"}}]

这说明我们的应用frontend和backend已经将trace信息写入kafka成功了！

在Zipkin的Web界面中，也能查询到这次跟踪信息

在zipkin的控制台，我们也看到跟Kafka相关的类ConsumerFetcherThread启动，我们在后续专门分析zipkin的源代码再来看看这个类。

2017-11-17 11:25:00.477  INFO 9292 --- [49-8e18eab0-0-1] kafka.consumer.ConsumerFetcherThread     : [ConsumerFetcherThread-zipkin_LT290-1510889099649-8e18eab0-0-1], Starting
2017-11-17 11:25:00.482  INFO 9292 --- [r-finder-thread] kafka.consumer.ConsumerFetcherManager    : [ConsumerFetcherManager-1510889099800] Added fetcher for partitions ArrayBuffer([[zipkin,0], initOffset 0 to broker id:1,host:10.200.170.137,port:9091] )

KafkaSender

public abstract class KafkaSender extends Sender {
   
     
  public static Builder newBuilder() {
    // Settings below correspond to "Producer Configs"
    // http://kafka.apache.org/0102/documentation.html#producerconfigs
    Properties properties = new Properties();
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
        ByteArraySerializer.class.getName());
    properties.put(ProducerConfig.ACKS_CONFIG, "0");
    return new zipkin2.reporter.kafka11.AutoValue_KafkaSender.Builder()
        .encoding(Encoding.JSON)
        .properties(properties)
        .topic("zipkin")
        .overrides(Collections.EMPTY_MAP)
        .messageMaxBytes(1000000);
  }

  @Override public zipkin2.Call<Void> sendSpans(List<byte[]> encodedSpans) {
    if (closeCalled) throw new IllegalStateException("closed");
    byte[] message = encoder().encode(encodedSpans);
    return new KafkaCall(message);
  }

}

KafkaSender中通过KafkaProducer客户端来发送消息给Kafka，在newBuilder方法中，设置了一些默认值，比如topic默认为zipkin，编码默认用JSON，消息最大字节数1000000，还可以通过overrides来覆盖默认的配置来定制KafkaProducer。

在sendSpans方法中返回KafkaCall，这个对象的execute方法，在AsyncReporter中的flush方法中会被调用：

void flush(BufferNextMessage bundler) {
    // ...
    sender.sendSpans(nextMessage).execute();
    // ...
}

KafkaCall的父类BaseCall方法execute会调用doExecute，而在doExecute方法中使用了一个AwaitableCallback将KafkaProducer的异步发送消息的方法，强制转为了同步发送，这里也确实处理的比较优雅。

  class KafkaCall extends BaseCall<Void> { // KafkaFuture is not cancelable
    private final byte[] message;

    KafkaCall(byte[] message) {
      this.message = message;
    }

    @Override protected Void doExecute() throws IOException {
      final AwaitableCallback callback = new AwaitableCallback();
      get().send(new ProducerRecord<>(topic(), message), (metadata, exception) -> {
        if (exception == null) {
          callback.onSuccess(null);
        } else {
          callback.onError(exception);
        }
      });
      callback.await();
      return null;
    }

    @Override protected void doEnqueue(Callback<Void> callback) {
      get().send(new ProducerRecord<>(topic(), message), (metadata, exception) -> {
        if (exception == null) {
          callback.onSuccess(null);
        } else {
          callback.onError(exception);
        }
      });
    }

    @Override public Call<Void> clone() {
      return new KafkaCall(message);
    }
  }

这里还有一个知识点，get方法每次都会返回一个新的KafkaProducer，我在第一眼看到这段代码时也曾怀疑，难道这里没有性能问题？
原来这里用到了google的插件autovalue里的标签@Memoized，结合@AutoValue标签，它会在自动生成的类里，给我们添加一些代码，可以看到get方法里作了一层缓存，所以我们的担心是没有必要的

  @Memoized KafkaProducer<byte[], byte[]> get() {
    KafkaProducer<byte[], byte[]> result = new KafkaProducer<>(properties());
    provisioned = true;
    return result;
  }

AutoValue_KafkaSender

final class AutoValue_KafkaSender extends $AutoValue_KafkaSender {
  private volatile KafkaProducer<byte[], byte[]> get;

  AutoValue_KafkaSender(Encoding encoding$, int messageMaxBytes$, BytesMessageEncoder encoder$,
      String topic$, Properties properties$) {
    super(encoding$, messageMaxBytes$, encoder$, topic$, properties$);
  }

  @Override
  KafkaProducer<byte[], byte[]> get() {
    if (get == null) {
      synchronized (this) {
        if (get == null) {
          get = super.get();
          if (get == null) {
            throw new NullPointerException("get() cannot return null");
          }
        }
      }
    }
    return get;
  }
}

KafkaCollector

我们再来看下Zipkin中的KafkaCollector，我们打开zipkin-server的源代码，在目录resources/zipkin-server-shared.yml文件中，发现关于kafka的配置片段
而我们在本文前面使用–KAFKA_ZOOKEEPER启动了zipkin，将kafka的zookeeper参数传递给了KafkaServer的main方法，也就是说，我们制定了zipkin.collector.kafka.zookeeper的值为localhost:2181

java -jar zipkin-server-2.2.1-exec.jar --KAFKA_ZOOKEEPER=localhost:2181

zipkin-server-shared.yml

zipkin:
  collector:
    kafka:
      ZooKeeper host string, comma-separated host:port value.
      zookeeper: ${
   
     KAFKA_ZOOKEEPER:}
      Name of topic to poll for spans
      topic: ${
   
     KAFKA_TOPIC:zipkin}
      Consumer group this process is consuming on behalf of.
      group-id: ${
   
     KAFKA_GROUP_ID:zipkin}
      Count of consumer threads consuming the topic
      streams: ${
   
     KAFKA_STREAMS:1}
      Maximum size of a message containing spans in bytes
      max-message-size: ${
   
     KAFKA_MAX_MESSAGE_SIZE:1048576}

在pom.xml中，有如下依赖

    <!-- Kafka Collector -->
    <dependency>
      <groupId>${project.groupId}</groupId>
      <artifactId>zipkin-autoconfigure-collector-kafka</artifactId>
      <optional>true</optional>
    </dependency>

ZipkinKafkaCollectorAutoConfiguration

我们找到zipkin-autoconfigure/collector-kafka的ZipkinKafkaCollectorAutoConfiguration类，使用了@Conditional注解，当KafkaZooKeeperSetCondition条件满足时，ZipkinKafkaCollectorAutoConfiguration类会被SpringBoot加载。当加载时，会配置KafkaCollector到spring容器中。

@Configuration
@EnableConfigurationProperties(ZipkinKafkaCollectorProperties.class)
@Conditional(KafkaZooKeeperSetCondition.class)
public class ZipkinKafkaCollectorAutoConfiguration {
   
     

  /**
   * This launches a thread to run start. This prevents a several second hang, or worse crash if
   * zookeeper isn't running, yet.
   */
  @Bean KafkaCollector kafka(ZipkinKafkaCollectorProperties kafka, CollectorSampler sampler,
      CollectorMetrics metrics, StorageComponent storage) {
    final KafkaCollector result =
        kafka.toBuilder().sampler(sampler).metrics(metrics).storage(storage).build();

    // don't use @Bean(initMethod = "start") as it can crash the process if zookeeper is down
    Thread start = new Thread("start " + result.getClass().getSimpleName()) {
      @Override public void run() {
        result.start();
      }
    };
    start.setDaemon(true);
    start.start();

    return result;
  }
}

KafkaZooKeeperSetCondition

KafkaZooKeeperSetCondition继承了SpringBootCondition，实现了getMatchOutcome方法，当上下文的环境变量中有配置zipkin.collector.kafka.zookeeper的时候，则条件满足，即ZipkinKafkaCollectorAutoConfiguration会被加载

final class KafkaZooKeeperSetCondition extends SpringBootCondition {
  static final String PROPERTY_NAME = "zipkin.collector.kafka.zookeeper";

  @Override
  public ConditionOutcome getMatchOutcome(ConditionContext context, AnnotatedTypeMetadata a) {
    String kafkaZookeeper = context.getEnvironment().getProperty(PROPERTY_NAME);
    return kafkaZookeeper == null || kafkaZookeeper.isEmpty() ?
        ConditionOutcome.noMatch(PROPERTY_NAME + " isn't set") :
        ConditionOutcome.match();
  }
}

在ZipkinKafkaCollectorAutoConfiguration中，启动了一个守护线程来运行KafkaCollector的start方法，避免zookeeper连不上，阻塞zipkin的启动过程。

public final class KafkaCollector implements CollectorComponent {
   
     
  final LazyConnector connector;
  final LazyStreams streams;

  KafkaCollector(Builder builder) {
    connector = new LazyConnector(builder);
    streams = new LazyStreams(builder, connector);
  }

  @Override public KafkaCollector start() {
    connector.get();
    streams.get();
    return this;
  }
}

KafkaCollector中初始化了两个对象，LazyConnector，和LazyStreams，在start方法中调用了2个对象的get方法

LazyConnector

LazyConnector继承了Lazy，当get方法被调用的时候，compute方法会被调用

  static final class LazyConnector extends LazyCloseable<ZookeeperConsumerConnector> {

    final ConsumerConfig config;

    LazyConnector(Builder builder) {
      this.config = new ConsumerConfig(builder.properties);
    }

    @Override protected ZookeeperConsumerConnector compute() {
      return (ZookeeperConsumerConnector) createJavaConsumerConnector(config);
    }

    @Override
    public void close() {
      ZookeeperConsumerConnector maybeNull = maybeNull();
      if (maybeNull != null) maybeNull.shutdown();
    }
  }

Lazy的get方法中，使用了典型的懒汉式单例模式，并使用了double-check，方式多线程构造多个实例，而真正构造对象是委派给compute方法

public abstract class Lazy<T> {
   
     

  volatile T instance = null;

  /** Remembers the result, if the operation completed unexceptionally. */
  protected abstract T compute();

  /** Returns the same value, computing as necessary */
  public final T get() {
    T result = instance;
    if (result == null) {
      synchronized (this) {
        result = instance;
        if (result == null) {
          instance = result = tryCompute();
        }
      }
    }
    return result;
  }

  /**
   * This is called in a synchronized block when the value to memorize hasn't yet been computed.
   *
   * <p>Extracted only for LazyCloseable, hence package protection.
   */
  T tryCompute() {
    return compute();
  }
}

在LazyConnector的compute方法中根据ConsumerConfig构造出了ZookeeperConsumerConnector，这个是kafka 0.8版本一种重要的对象，基于zookeeper的ConsumerConnector。

LazyStreams

在LazyStreams的compute中，新建了一个线程池，线程池大小可以由参数streams（即zipkin.collector.kafka.streams）来指定，默认为一个线程的线程池。
然后通过topicCountMap设置zipkin的kafka消费使用的线程数，再使用ZookeeperConsumerConnector的createMessageStreams方法来创建KafkaStream，然后使用线程池执行KafkaStreamProcessor。

  static final class LazyStreams extends LazyCloseable<ExecutorService> {
    final int streams;
    final String topic;
    final Collector collector;
    final CollectorMetrics metrics;
    final LazyCloseable<ZookeeperConsumerConnector> connector;
    final AtomicReference<CheckResult> failure = new AtomicReference<>();

    LazyStreams(Builder builder, LazyCloseable<ZookeeperConsumerConnector> connector) {
      this.streams = builder.streams;
      this.topic = builder.topic;
      this.collector = builder.delegate.build();
      this.metrics = builder.metrics;
      this.connector = connector;
    }

    @Override protected ExecutorService compute() {
      ExecutorService pool = streams == 1
          ? Executors.newSingleThreadExecutor()
          : Executors.newFixedThreadPool(streams);

      Map<String, Integer> topicCountMap = new LinkedHashMap<>(1);
      topicCountMap.put(topic, streams);

      for (KafkaStream<byte[], byte[]> stream : connector.get().createMessageStreams(topicCountMap)
          .get(topic)) {
        pool.execute(guardFailures(new KafkaStreamProcessor(stream, collector, metrics)));
      }
      return pool;
    }

    Runnable guardFailures(final Runnable delegate) {
      return () -> {
        try {
          delegate.run();
        } catch (RuntimeException e) {
          failure.set(CheckResult.failed(e));
        }
      };
    }

    @Override
    public void close() {
      ExecutorService maybeNull = maybeNull();
      if (maybeNull != null) maybeNull.shutdown();
    }
  }

KafkaStreamProcessor

在KafkaStreamProcessor的run方法中，迭代stream对象，取出获得的流数据，然后调用Collector的acceptSpans方法，即使用storage组件来接收并存储span数据。

final class KafkaStreamProcessor implements Runnable {
  final KafkaStream<byte[], byte[]> stream;
  final Collector collector;
  final CollectorMetrics metrics;

  KafkaStreamProcessor(
      KafkaStream<byte[], byte[]> stream, Collector collector, CollectorMetrics metrics) {
    this.stream = stream;
    this.collector = collector;
    this.metrics = metrics;
  }

  @Override
  public void run() {
    ConsumerIterator<byte[], byte[]> messages = stream.iterator();
    while (messages.hasNext()) {
      byte[] bytes = messages.next().message();
      metrics.incrementMessages();

      if (bytes.length == 0) {
        metrics.incrementMessagesDropped();
        continue;
      }

      // If we received legacy single-span encoding, decode it into a singleton list
      if (bytes[0] <= 16 && bytes[0] != 12 /* thrift, but not a list */) {
        try {
          metrics.incrementBytes(bytes.length);
          Span span = SpanDecoder.THRIFT_DECODER.readSpan(bytes);
          collector.accept(Collections.singletonList(span), NOOP);
        } catch (RuntimeException e) {
          metrics.incrementMessagesDropped();
        }
      } else {
        collector.acceptSpans(bytes, DETECTING_DECODER, NOOP);
      }
    }
  }
}

这里的kafka消费方式还是kafka0.8版本的，如果你想用kafka0.10+的版本，可以更改zipkin-server的pom，将collector-kafka10加入到依赖中，其原理跟kafka0.8的差不多，此处不再展开分析了。

<!-- Kafka10 Collector -->
<dependency>
  <groupId>io.zipkin.java</groupId>
  <artifactId>zipkin-autoconfigure-collector-kafka10</artifactId>
  <optional>true</optional>
</dependency>

<dependency>
  <groupId>io.zipkin.java</groupId>
  <artifactId>zipkin-collector-kafka10</artifactId>
</dependency>

在生产环境中，我们可以将zipkin的日志收集器改为kafka来提高系统的吞吐量，而且也可以让客户端和zipkin服务端解耦，客户端将不依赖zipkin服务端，只依赖kafka集群。

当然我们也可以将zipkin的collector替换为RabbitMQ来提高日志收集的效率，zipkin对scribe也作了支持，这里就不展开篇幅细说了。