15.8. 用于追踪的生产者和消费者

先决条件

• 您已为 客户端 初始化追踪。

  您可以通过在项目中添加追踪 JAR 作为依赖项来启用制作者和使用者应用。

• 要使用 decorator 模式，请创建修改后的制作者或消费者实例来发送或接收消息。

  您传递了原始 KafkaProducer 或 KafkaConsumer 类。

  OpenTelemetry 的 decorator 检测示例

  // Producer instance
  Producer < String, String > op = new KafkaProducer < > (
      configs,
      new StringSerializer(),
      new StringSerializer()
      );
      Producer < String, String > producer = tracing.wrap(op);
  KafkaTracing tracing = KafkaTracing.create(GlobalOpenTelemetry.get());
  producer.send(...);
  
  //consumer instance
  Consumer<String, String> oc = new KafkaConsumer<>(
      configs,
      new StringDeserializer(),
      new StringDeserializer()
      );
      Consumer<String, String> consumer = tracing.wrap(oc);
  consumer.subscribe(Collections.singleton("mytopic"));
  ConsumerRecords<Integer, String> records = consumer.poll(1000);
  ConsumerRecord<Integer, String> record = ...
  SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer);

  OpenTracing 的 decorator 检测示例

  //producer instance
  KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps);
  TracingKafkaProducer<Integer, String> tracingProducer = new TracingKafkaProducer<>(producer, tracer);
  TracingKafkaProducer.send(...)
  
  //consumer instance
  KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);
  TracingKafkaConsumer<Integer, String> tracingConsumer = new TracingKafkaConsumer<>(consumer, tracer);
  tracingConsumer.subscribe(Collections.singletonList("mytopic"));
  ConsumerRecords<Integer, String> records = tracingConsumer.poll(1000);
  ConsumerRecord<Integer, String> record = ...
  SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer);

• 要使用拦截器，请在制作者或消费者配置中设置拦截器类。

  您以通常的方式使用 KafkaProducer 和 KafkaConsumer 类。TracingProducerInterceptor 和 TracingConsumerInterceptor interceptor 类负责追踪功能。

  使用拦截器的制作者配置示例

  senderProps.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
      TracingProducerInterceptor.class.getName());
  
  KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps);
  producer.send(...);

  使用拦截器的使用者配置示例

  consumerProps.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
      TracingConsumerInterceptor.class.getName());
  
  KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);
  consumer.subscribe(Collections.singletonList("messages"));
  ConsumerRecords<Integer, String> records = consumer.poll(1000);
  ConsumerRecord<Integer, String> record = ...
  SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer);