说明
这里以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，因为专业的人已经帮我们造好轮子了，就不要自己再去造轮子了
