sleuth or 自定义追踪kafka消息

说明

这里以spring-kafka为例子，trace框架用的是spring-sleuth
对应的版本号见如下

     <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-starter-sleuth</artifactId>
                <version>3.1.3</version>
            </dependency>

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

第一步

引入以上依赖，然后配置自己的kafka producer以及consumer

第二步

检查kafka server版本是否支持header传递，

第三步

配置trace 日志格式

第四步

进行测试，可以发现生产时后打印的日志与消费时候打印的日志的traceId为同一个

注意点
因为这里使用的是spring-kafka，所以当我们在创建生产者和消费者时，必须使用spring的方式，不要自己手动去配置否则将不能跟踪起来；如果非要手动创建，那么建议不要使用spring-kafka，用apache-kafka，然后用TraceingKafkaProducer进行包装就ok。

原理说明
实现原理其实很简单，就是用了kafka的消息头，发送的时候将traceId放入，消息头，消费时候再拿出来
同理，自己也可以简单实现，利用拦截器，在发送与消费的时候进行设置。如下是简单的demo

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.Header;
import org.slf4j.MDC;

import java.util.Arrays;
import java.util.Map;

/**
 * className:TracingConsumerMethodInterceptor
 * description: consumer 自定义拦截器实现trace demo 目前没使用
 *
 * @author: david
 * date: 2022/9/7 09:48
 */


@Slf4j
public class TracingConsumerMethodInterceptor implements ConsumerInterceptor<String,String> {


    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> consumerRecords) {
        consumerRecords.forEach(x->{
            Header[] headers = x.headers().toArray();
            Arrays.asList(headers)
                    .forEach(s->{
                        if(s.key().equals("trace_id")){
                            MDC.put(s.key(),new String(s.value()));
                        }
                    });
        });
        return consumerRecords;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> map) {

    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.MDC;
import org.springframework.util.StringUtils;

import java.nio.charset.StandardCharsets;
import java.util.Map;

/**
 * className:TracingProducerMethodInterceptor
 * description:producer 自定义拦截器实现trace demo 目前没使用
 *
 * @author: david
 * date: 2022/9/7 09:48
 */


@Slf4j
public class TracingProducerMethodInterceptor implements ProducerInterceptor<String,String> {


    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
        String traceId = MDC.get("trace_id");
        if(StringUtils.isEmpty(traceId)){
            traceId="222222";
        }
        MDC.put("trace_id",traceId);
        producerRecord.headers().add("trace_id",traceId.getBytes(StandardCharsets.UTF_8));
        return producerRecord;
    }

    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {

    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}

然后在创建producer以及consumer时注入以上拦截器，便可以实现
当然最好使用spring-kafka，因为专业的人已经帮我们造好轮子了，就不要自己再去造轮子了