8. Kafka分布式链路追踪实现设计-六虎

前语

本文将从Kafka供给的扩展机制出发，结合Opentracing和Jaeger，完成Kafka的分布式链路追寻。

kafka-clients版别：3.1.2
spring-kafka版别：2.8.11

正文

一. Kafka的阻拦器和监听器

先看一下Kafka的阻拦器和监听器的接口界说。

首先是阻拦器，分为出产者阻拦器ProducerInterceptor和顾客阻拦器ConsumerInterceptor，接口界说如下。

public interface ProducerInterceptor<K, V> extends Configurable {
    // 音讯被发送前履行
    ProducerRecord<K, V> onSend(ProducerRecord<K, V> record);
    // 音讯发送出现反常或许发送后触发回调时履行
    void onAcknowledgement(RecordMetadata metadata, Exception exception);
    void close();
}

public interface ConsumerInterceptor<K, V> extends Configurable, AutoCloseable {
    // 音讯拉取回来前履行
    ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records);
    // 提交offset后履行
    void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets);
    void close();
}

再看一下监听器的接口界说，如下所示。

public interface ProducerListener<K, V> {
    // 音讯发送成功后触发回调时履行
    default void onSuccess(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata) {
    }
    // 音讯发送有反常触发回调时履行
    default void onError(ProducerRecord<K, V> producerRecord, @Nullable RecordMetadata recordMetadata,
                         Exception exception) {
    }
}

那么出产者出产音讯和顾客拉取音讯这两个动作，结合阻拦器和监听器，一个简单的图示图下。

二. Kafka分布式链路追寻Span模型规划

Kafka作为音讯中间件，支持音讯的push和pull，这姿态的信息传递方法，有别于运用RestTemplate恳求传递音讯，可是实质还是将音讯传递到了下流，那么咱们稍加笼统，就能够得到如下的Span模型。

其实便是在出产者这里发送音讯时创立代表下流的Span，然后将Span经过Record的Header来传递给到下流，这姿态全体的思路就和咱们运用RestTemplate恳求下流时，把Span放到HTTP恳求头里传递给下流是相同的。

三. Kafka出产者链路追寻规划和完成

1. Kafka出产者链路追寻规划

初步看好像能够根据出产者阻拦器来完成，例如在onSend() 办法中创立出来Span并加到ProducerRecord的Header中，然后在onAcknowledgement() 办法中调用Span的finish() 办法来记载Span，可是实际是不能根据出产者阻拦器来完成的，理由如下。

在onSend() 和onAcknowledgement() 办法中，入参都只要发送的音讯，而咱们记载Span时需求的一些其它信息，很难拿到；
没有办法把Tracer和装修器等对象设置给出产者阻拦器，这是因为出产者阻拦器的初始化是在KafkaProducer中读取到阻拦器的全限定名然后再经过反射的方法创立出来的，咱们很难把Tracer和装修器等对象设置给出产者阻拦器。

鉴于上述的问题，咱们就不能根据出产者阻拦器来完成Kafka出产者链路追寻，有一个很好的替代的思路是，咱们在KafkaTemplate之上再包装一层，作为咱们自己的增强的KafkaTemplate，这里命名为HoneyKafkaTemplate，然后咱们再界说自己的阻拦器接口HoneyKafkaProducerInterceptor，当运用HoneyKafkaTemplate来发送音讯时，会先进入到HoneyKafkaProducerInterceptor，创立Span和记载Span的功用都写在HoneyKafkaProducerInterceptor中，终究发送音讯时，还是将这个发送的动作委派给原生的KafkaTemplate，全体的结构如下所示。

2. Kafka出产者链路追寻完成

首先咱们完成HoneyKafkaTemplate来替代原生的KafkaTemplate，HoneyKafkaTemplate如下所示。

public class HoneyKafkaTemplate<K, V> extends KafkaTemplate<K, V> {
    public HoneyKafkaTemplate(ProducerFactory<K, V> producerFactory) {
        super(producerFactory);
    }
    public HoneyKafkaTemplate(ProducerFactory<K, V> producerFactory, Map<String, Object> configOverrides) {
        super(producerFactory, configOverrides);
    }
    public HoneyKafkaTemplate(ProducerFactory<K, V> producerFactory, boolean autoFlush) {
        super(producerFactory, autoFlush);
    }
    public HoneyKafkaTemplate(ProducerFactory<K, V> producerFactory, boolean autoFlush, Map<String, Object> configOverrides) {
        super(producerFactory, autoFlush, configOverrides);
    }
}

然后供给一个主动安装的装备类HoneyKafkaTemplateConfig，如下所示。

@Configuration
@AutoConfigureBefore(KafkaAutoConfiguration.class)
public class HoneyKafkaTemplateConfig {
    @Bean
    public HoneyKafkaTemplate<?, ?> kafkaTemplate(ProducerFactory<Object, Object> kafkaProducerFactory) {
        return new HoneyKafkaTemplate<>(kafkaProducerFactory);
    }
}

因为KafkaAutoConfiguration在注册KafkaTemplate时，添加了@ConditionalOnMissingBean(KafkaTemplate.class)，所以只需求保证咱们的HoneyKafkaTemplateConfig在KafkaAutoConfiguration之前被处理，那么咱们注册的HoneyKafkaTemplate就能替代原生的KafkaTemplate来到达以假乱真的作用。

终究要在spring.factories文件中指定咱们的主动安装类以及在pom文件中添加spring-kafka依靠，如下所示。

org.springframework.boot.autoconfigure.EnableAutoConfiguration=
 com.honey.tracing.config.HoneyTracingConfig,
 com.honey.tracing.config.HoneyTracingFilterConfig,
 com.honey.tracing.config.HoneyRestTemplateTracingConfig,
 com.honey.tracing.config.HoneyKafkaTemplateConfig

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <scope>provided</scope>
</dependency>

现在再给出自界说阻拦器HoneyKafkaProducerInterceptor的界说，如下所示。

public interface HoneyKafkaProducerInterceptor<K, V> {
    /**
     * 阻拦Kafka出产者的音讯发送。
     *
     * @param producerRecord 出产者发送的音讯。
     */
    ListenableFuture<SendResult<K, V>> intercept(
            ProducerRecord<K, V> producerRecord,
            HoneyKafkaProducerInterceptorChain<K, V> kafkaProducerInterceptorChain);
}

咱们再供给一个完成类，如下所示。

public class HoneyKafkaProducerTracingInterceptor<K, V> implements HoneyKafkaProducerInterceptor<K, V> {
    private final Tracer tracer;
    private final List<HoneyKafkaTracingProducerDecorator<K, V>> kafkaTracingProducerDecorators;
    public HoneyKafkaProducerTracingInterceptor(Tracer tracer, List<HoneyKafkaTracingProducerDecorator<K, V>> kafkaTracingProducerDecorators) {
        this.tracer = tracer;
        this.kafkaTracingProducerDecorators = kafkaTracingProducerDecorators;
    }
    @Override
    public ListenableFuture<SendResult<K, V>> intercept(ProducerRecord<K, V> producerRecord,
                                                        HoneyKafkaProducerInterceptorChain<K, V> kafkaProducerInterceptorChain) {
        if (tracer.activeSpan() == null) {
            return kafkaProducerInterceptorChain.intercept(producerRecord);
        }
        // 生成Kafka出产者对应的Span
        // 类似于RestTemplate调用前的Span
        Span span = tracer.buildSpan(HONEY_KAFKA_NAME)
                .withTag(Tags.SPAN_KIND.getKey(), Tags.SPAN_KIND_CLIENT)
                .start();
        for (HoneyKafkaTracingProducerDecorator<K, V> kafkaTracingProducerDecorator : kafkaTracingProducerDecorators) {
            try {
                kafkaTracingProducerDecorator.onSend(span, producerRecord);
            } catch (Exception e) {
                // do nothing
            }
        }
        ListenableFuture<SendResult<K, V>> result;
        try (Scope scope = tracer.activateSpan(span)) {
            // 设置Kafka服务端的host
            String hostString = kafkaProducerInterceptorChain.getHoneyKafkaTemplate().getProducerFactory()
                    .getConfigurationProperties().get(BOOTSTRAP_SERVERS_CONFIG).toString();
            span.setTag(FIELD_HOST, hostString.substring(1, hostString.length() - 1));
            // host需求传递给顾客
            span.setBaggageItem(FIELD_HOST, hostString.substring(1, hostString.length() - 1));
            // 把SpanContext注入到ProducerRecord的Headers中
            tracer.inject(span.context(), Format.Builtin.HTTP_HEADERS, new HoneyKafkaCarrier(producerRecord.headers()));
            try {
                result = kafkaProducerInterceptorChain.intercept(producerRecord);
            } catch (Exception e1) {
                for (HoneyKafkaTracingProducerDecorator<K, V> kafkaTracingProducerDecorator : kafkaTracingProducerDecorators) {
                    try {
                        kafkaTracingProducerDecorator.onError(span, producerRecord);
                    } catch (Exception e2) {
                        // do nothing
                    }
                }
                throw e1;
            }
            for (HoneyKafkaTracingProducerDecorator<K, V> kafkaTracingProducerDecorator : kafkaTracingProducerDecorators) {
                try {
                    kafkaTracingProducerDecorator.onSuccess(span, producerRecord);
                } catch (Exception e) {
                    // do nothing
                }
            }
        } finally {
            span.finish();
            tracer.activeSpan().log(RequestStackUtil.assembleRequestStack((JaegerSpan) span));
        }
        return result;
    }
}

和之前完成的RestTemplate的阻拦器逻辑几乎是一模相同，区别就在于一个是Span放在HTTP恳求头，一个是Span放在音讯的Header中，以及咱们在Inject时运用了一个自界说的HoneyKafkaCarrier，如下所示。

public class HoneyKafkaCarrier implements TextMap {
    private final Headers headers;
    public HoneyKafkaCarrier(Headers headers) {
        this.headers = headers;
    }
    @NotNull
    @Override
    public Iterator<Map.Entry<String, String>> iterator() {
        Map<String, String> headerMap = new HashMap<>();
        for (Header header : headers) {
            headerMap.put(header.key(), new String(header.value()));
        }
        return headerMap.entrySet().iterator();
    }
    @Override
    public void put(String key, String value) {
        headers.add(key, value.getBytes());
    }
}

特别注意到HoneyKafkaProducerInterceptorChain，咱们将所有的HoneyKafkaProducerInterceptor创立为了一个连接器链，如下所示。

public class HoneyKafkaProducerInterceptorChain<K, V> {
    private final Iterator<HoneyKafkaProducerInterceptor<K, V>> kafkaProducerInterceptors;
    private final HoneyKafkaTemplate<K, V> honeyKafkaTemplate;
    public HoneyKafkaProducerInterceptorChain(Iterator<HoneyKafkaProducerInterceptor<K, V>> kafkaProducerInterceptors,
                                              HoneyKafkaTemplate<K, V> honeyKafkaTemplate) {
        this.kafkaProducerInterceptors = kafkaProducerInterceptors;
        this.honeyKafkaTemplate = honeyKafkaTemplate;
    }
    public HoneyKafkaTemplate<K, V> getHoneyKafkaTemplate() {
        return honeyKafkaTemplate;
    }
    public ListenableFuture<SendResult<K, V>> intercept(ProducerRecord<K, V> producerRecord) {
        if (kafkaProducerInterceptors.hasNext()) {
            // 阻拦器没履行完则先履行阻拦器
            return kafkaProducerInterceptors.next().intercept(producerRecord, this);
        }
        // 阻拦器悉数履行完后才发送音讯
        return honeyKafkaTemplate.actuallySend(producerRecord);
    }
}

意图如下。

支持后续添加更多的阻拦器；
让连接器链持有HoneyKafkaTemplate从而所有阻拦器都持有了HoneyKafkaTemplate。

至于装修器，咱们暂时就界说一个空接口出来，后边按需添加完成类，如下所示。

/**
 * Kafka出产者链路追寻装修器。
 */
public interface HoneyKafkaTracingProducerDecorator<K, V> {
    void onSend(Span span, ProducerRecord<K, V> producerRecord);
    void onSuccess(Span span, ProducerRecord<K, V> producerRecord);
    void onError(Span span, ProducerRecord<K, V> producerRecord);
}

现在已经有阻拦器链了，接下来要做的事情便是把阻拦器链给到咱们的HoneyKafkaTemplate，如下所示。

public class HoneyKafkaTemplate<K, V> extends KafkaTemplate<K, V> {
    private List<HoneyKafkaProducerInterceptor<K, V>> kafkaProducerInterceptors;
    public HoneyKafkaTemplate(ProducerFactory<K, V> producerFactory,
                              List<HoneyKafkaProducerInterceptor<K, V>> kafkaProducerInterceptors) {
        super(producerFactory);
        this.kafkaProducerInterceptors = kafkaProducerInterceptors;
    }
    public HoneyKafkaTemplate(ProducerFactory<K, V> producerFactory, Map<String, Object> configOverrides) {
        super(producerFactory, configOverrides);
    }
    public HoneyKafkaTemplate(ProducerFactory<K, V> producerFactory, boolean autoFlush) {
        super(producerFactory, autoFlush);
    }
    public HoneyKafkaTemplate(ProducerFactory<K, V> producerFactory, boolean autoFlush, Map<String, Object> configOverrides) {
        super(producerFactory, autoFlush, configOverrides);
    }
    public ListenableFuture<SendResult<K, V>> actuallySend(ProducerRecord<K, V> producerRecord) {
        return super.doSend(producerRecord);
    }
    @NotNull
    @Override
    public ListenableFuture<SendResult<K, V>> doSend(@NotNull ProducerRecord<K, V> producerRecord) {
        HoneyKafkaProducerInterceptorChain<K, V> kafkaProducerInterceptorChain
                = new HoneyKafkaProducerInterceptorChain<>(kafkaProducerInterceptors.iterator(), this);
        return kafkaProducerInterceptorChain.intercept(producerRecord);
    }
}

这样在运用HoneyKafkaTemplate来发送音讯时，能够先让阻拦器链履行，然后再把音讯委派给原生KafkaTemplate来发送。

现在需求修正注册HoneyKafkaTemplate的主动安装类，如下所示。

@Configuration
@ConditionalOnBean(KafkaTemplate.class)
@AutoConfigureBefore(KafkaAutoConfiguration.class)
public class HoneyKafkaTemplateConfig {
    @Bean
    public HoneyKafkaTemplate<?, ?> kafkaTemplate(ProducerFactory<Object, Object> kafkaProducerFactory,
                                                  List<HoneyKafkaProducerInterceptor<Object, Object>> kafkaProducerInterceptors) {
        return new HoneyKafkaTemplate<>(kafkaProducerFactory, kafkaProducerInterceptors);
    }
}

即在创立HoneyKafkaTemplate时需求把注册到容器中的HoneyKafkaProducerInterceptor设置进去。一起还要为咱们的阻拦器的完成HoneyKafkaProducerTracingInterceptor供给一个主动安装类，如下所示。

@Configuration
@ConditionalOnBean(KafkaTemplate.class)
@AutoConfigureAfter(HoneyTracingConfig.class)
public class HoneyKafkaTracingConfig {
    @Bean
    public HoneyKafkaProducerInterceptor kafkaProducerTracingInterceptor(
            Tracer tracer, List<HoneyKafkaTracingProducerDecorator<Object, Object>> kafkaTracingProducerDecorators) {
        return new HoneyKafkaProducerTracingInterceptor<>(tracer, kafkaTracingProducerDecorators);
    }
}

对应修正spring.factories文件如下所示。

org.springframework.boot.autoconfigure.EnableAutoConfiguration=
 com.honey.tracing.config.HoneyTracingConfig,
 com.honey.tracing.config.HoneyTracingFilterConfig,
 com.honey.tracing.config.HoneyRestTemplateTracingConfig,
 com.honey.tracing.config.HoneyKafkaTemplateConfig,
 com.honey.tracing.config.HoneyKafkaTracingConfig

至此咱们的具有分布式链路追寻功用的HoneyKafkaTemplate就能够运用啦，终究给出上述运用到的常量类，如下所示。

public class CommonConstants {
    public static final double DEFAULT_SAMPLE_RATE = 1.0;
    public static final String HONEY_TRACER_NAME = "HoneyTracer";
    public static final String HONEY_REST_TEMPLATE_NAME = "HoneyRestTemplate";
    public static final String HONEY_KAFKA_NAME = "HoneyKafka";
    public static final String FIELD_HOST = "host";
    public static final String FIELD_API = "api";
    public static final String FIELD_HTTP_CODE = "httpCode";
    public static final String FIELD_SUB_SPAN_ID = "subSpanId";
    public static final String FIELD_SUB_HTTP_CODE = "subHttpCode";
    public static final String FIELD_SUB_TIMESTAMP = "subTimestamp";
    public static final String FIELD_SUB_DURATION = "subDuration";
    public static final String FIELD_SUB_HOST = "subHost";
    public static final String HOST_PATTERN_STR = "(?<=(https://|http://)).*?(?=/)";
    public static final String SLASH = "/";
    public static final String LOG_EVENT_KIND = "logEventKind";
    public static final String LOG_EVENT_KIND_REQUEST_STACK = "requestStack";
}

四. Kafka顾客链路追寻规划和完成

1. Kafka顾客链路追寻规划

同样的，Kafka顾客阻拦器也不适合来完成分布式链路追寻，理由和第三节第1末节中基本共同。

通常在Spring中完成Kafka音讯消费时，咱们运用@KafkaListener注解，被该注解修饰的办法用于消费音讯并处理，入参是ConsumerRecord，所以咱们能够挑选供给切面来切这些被@KafkaListener注解修饰的办法并在办法履行前后操作Span，一起为了和@KafkaListener注解解耦，咱们能够专门界说一个注解来作为咱们的切点，那么现在的音讯消费流程就成下面这样了。

2. Kafka顾客链路追寻完成

首先界说作为切点的注解，如下所示。

/**
 * Kafka顾客办法运用该注解。
 */
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface HoneyKafkaTracing {
}

对应的切面如下所示。

@Aspect
public class HoneyKafkaConsumerTracingAspect {
    private final HoneyKafkaConsumerInterceptor kafkaConsumerTracingInterceptor;
    public HoneyKafkaConsumerTracingAspect(HoneyKafkaConsumerInterceptor kafkaConsumerTracingInterceptor) {
        this.kafkaConsumerTracingInterceptor = kafkaConsumerTracingInterceptor;
    }
    @Pointcut("@annotation(com.honey.tracing.kafka.consumer.interceptor.HoneyKafkaTracing)")
    private void kafkaTracing() {
    }
    @Around("kafkaTracing()")
    public Object intercept(ProceedingJoinPoint joinPoint) throws Throwable {
        return kafkaConsumerTracingInterceptor.intercept(joinPoint);
    }
}

在盘绕告诉中，会调用到咱们自界说的Kafka顾客阻拦器，接口界说如下。

public interface HoneyKafkaConsumerInterceptor {
    Object intercept(ProceedingJoinPoint joinPoint) throws Throwable;
}

具体的完成类，如下所示。

public class HoneyKafkaConsumerTracingInterceptor<K, V> implements HoneyKafkaConsumerInterceptor {
    private final Tracer tracer;
    private final List<HoneyKafkaTracingConsumerDecorator<K, V>> kafkaTracingConsumerDecorators;
    public HoneyKafkaConsumerTracingInterceptor(
            Tracer tracer, List<HoneyKafkaTracingConsumerDecorator<K, V>> kafkaTracingConsumerDecorators) {
        this.tracer = tracer;
        this.kafkaTracingConsumerDecorators = kafkaTracingConsumerDecorators;
    }
    @Override
    public Object intercept(ProceedingJoinPoint joinPoint) throws Throwable {
        ConsumerRecord consumerRecord = null;
        // 找到Kafka音讯
        Object[] args = joinPoint.getArgs();
        for (Object arg : args) {
            if (arg instanceof ConsumerRecord) {
                consumerRecord = (ConsumerRecord) arg;
            }
        }
        if (consumerRecord == null) {
            // 没有获取到Kafka音讯则不处理链路
            return joinPoint.proceed();
        }
        SpanContext extractSpanContext = tracer.extract(Format.Builtin.HTTP_HEADERS,
                new HoneyKafkaCarrier(consumerRecord.headers()));
        Span span = tracer.buildSpan(HONEY_KAFKA_NAME)
                .asChildOf(extractSpanContext)
                .withTag(Tags.SPAN_KIND.getKey(), Tags.SPAN_KIND_SERVER)
                .start();
        span.setTag(FIELD_HOST, span.getBaggageItem(FIELD_HOST));
        for (HoneyKafkaTracingConsumerDecorator<K, V> kafkaTracingConsumerDecorator : kafkaTracingConsumerDecorators) {
            try {
                kafkaTracingConsumerDecorator.onReceive(span, consumerRecord);
            } catch (Exception e) {
                // do nothing
            }
        }
        Object result;
        try (Scope scope = tracer.activateSpan(span)) {
            try {
                result = joinPoint.proceed();
            } catch (Exception e1) {
                for (HoneyKafkaTracingConsumerDecorator<K, V> kafkaTracingConsumerDecorator : kafkaTracingConsumerDecorators) {
                    try {
                        kafkaTracingConsumerDecorator.onError(span, consumerRecord);
                    } catch (Exception e2) {
                        // do nothing
                    }
                }
                throw e1;
            }
            for (HoneyKafkaTracingConsumerDecorator<K, V> kafkaTracingConsumerDecorator : kafkaTracingConsumerDecorators) {
                try {
                    kafkaTracingConsumerDecorator.onFinished(span, consumerRecord);
                } catch (Exception e) {
                    // do nothing
                }
            }
        } finally {
            span.finish();
        }
        return result;
    }
}

同样的咱们也界说了一个装修器接口，如下所示。

/**
 * Kafka顾客链路追寻装修器。
 */
public interface HoneyKafkaTracingConsumerDecorator<K, V> {
    void onReceive(Span span, ConsumerRecord<K, V> consumerRecord);
    void onFinished(Span span, ConsumerRecord<K, V> consumerRecord);
    void onError(Span span, ConsumerRecord<K, V> consumerRecord);
}

然后需求修正一下主动安装类HoneyKafkaTracingConfig来将咱们自界说的Kafka顾客阻拦器和切面注册到容器中，修正如下。

@Configuration
@AutoConfigureAfter(HoneyTracingConfig.class)
public class HoneyKafkaTracingConfig {
    @Bean
    public HoneyKafkaProducerInterceptor kafkaProducerTracingInterceptor(
            Tracer tracer, List<HoneyKafkaTracingProducerDecorator<Object, Object>> kafkaTracingProducerDecorators) {
        return new HoneyKafkaProducerTracingInterceptor<>(tracer, kafkaTracingProducerDecorators);
    }
    @Bean
    public HoneyKafkaConsumerInterceptor honeyKafkaConsumerInterceptor(
            Tracer tracer, List<HoneyKafkaTracingConsumerDecorator<Object, Object>> kafkaTracingProducerDecorators) {
        return new HoneyKafkaConsumerTracingInterceptor(tracer, kafkaTracingProducerDecorators);
    }
    @Bean
    public HoneyKafkaConsumerTracingAspect honeyKafkaConsumerTracingAspect(
            HoneyKafkaConsumerInterceptor kafkaConsumerTracingInterceptor) {
        return new HoneyKafkaConsumerTracingAspect(kafkaConsumerTracingInterceptor);
    }
}

终究，因为运用到了切面，所以需求引进Spring的AOP的依靠，如下所示。

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-aop</artifactId>
</dependency>

五. 功用验证

现在运用一个简单的Kafka的例子来验证Kafka分布式链路追寻的功用。

咱们让example-service-1充任出产者，pom中引进Kafka依靠如下所示。

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

然后添加出产者的装备，如下所示。

spring:
  kafka:
    bootstrap-servers: 127.0.0.1:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

终究添加一个KafkaController来发送Kafka音讯，如下所示。

@RestController
public class KafkaController {
    private static final String TEST_TOPIC = "testTopic";
    private static final String TEST_MESSAGE = "testMessage";
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @GetMapping("/kafka/send")
    public void send(String url) {
        kafkaTemplate.send(TEST_TOPIC, TEST_MESSAGE);
    }
}

咱们再让example-service-2充任顾客，pom中引进Kafka依靠如下所示。

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

然后添加顾客的装备，如下所示。

spring:
  kafka:
    bootstrap-servers: 127.0.0.1:9092
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      group-id: kafka-tracing

终究添加一个KafkaService来消费音讯，如下所示。

@Service
public class KafkaService {
    private static final String TEST_TOPIC = "testTopic";
    @HoneyKafkaTracing
    @KafkaListener(topics = TEST_TOPIC)
    public void onMessage(ConsumerRecord<String, String> consumerRecord) {
        System.out.println(consumerRecord.value());
    }
}

终究example-service-1打印链路如下所示。

{
	"traceId": "904360d3ae85faa7ce5d698debda55fd",
	"spanId": "ce5d698debda55fd",
	"parentSpanId": "0000000000000000",
	"timestamp": "1708067308855",
	"duration": "306",
	"httpCode": "200",
	"host": "http://localhost:8080",
	"requestStacks": [{
		"subSpanId": "7cb4062ca0fa7cdd",
		"subHttpCode": "null",
		"subTimestamp": "1708067308876",
		"subDuration": "262",
		"subHost": "127.0.0.1:9092"
	}]
}

example-service-2打印链路如下所示。

{
	"traceId": "904360d3ae85faa7ce5d698debda55fd",
	"spanId": "7cb4062ca0fa7cdd",
	"parentSpanId": "ce5d698debda55fd",
	"timestamp": "1708067309171",
	"duration": "5",
	"httpCode": "null",
	"host": "127.0.0.1:9092",
	"requestStacks": []
}

可见链路信息经过Kafka音讯得到了传递。

总结

Kafka尽管是经过push和pull方法来传递音讯，可是咱们能够将这种方法笼统为出产者直接将音讯传递给到了顾客，传递介质便是Record，一起Record中预留了Header，咱们就能够将Span经过Header来传递，整个模式就和经过RestTemplate恳求下流一模相同了。

Kafka分布式链路追寻的完成，实质是阻拦Kafka的音讯发送和接纳，尽管Kafka供给了阻拦器，可是因为Kafka阻拦器的加载机制问题，运用起来并不是很方便，所以咱们挑选自界说阻拦器和切面的方法来阻拦Kafka的音讯发送和接纳，一起为了方便的运用咱们自界说的阻拦器，咱们还自界说了KafkaTemplate来对原生KafkaTemplate进行了包装，运用自界说KafkaTemplate发送音讯时，能够先在自界说KafkaTemplate中先应用自界说阻拦器的逻辑，然后真实发送音讯的动作就委派给原生KafkaTemplate。

本文中，honey-starter-tracing的工程目录结构如下所示。

测试demo的工程目录结构如下所示。

8. Kafka分布式链路追踪实现设计

前语

正文

一. Kafka的阻拦器和监听器

二. Kafka分布式链路追寻Span模型规划

三. Kafka出产者链路追寻规划和完成

1. Kafka出产者链路追寻规划

2. Kafka出产者链路追寻完成

四. Kafka顾客链路追寻规划和完成

1. Kafka顾客链路追寻规划

2. Kafka顾客链路追寻完成

五. 功用验证

总结

相关文章

javascript点击图片在弹出层显示大图

人工智能chatGPT注册流程及体验

千万不要用JSON.stringify()去实现深拷贝！有巨坑！！

关于Spring的两三事：万物之始—BeanDefinition

作者信息