主页
产品
Red Hat AMQ
2021.Q3
在 OpenShift 中使用 AMQ Streams
10.3. 使用 tracers 强制 Kafka 客户端

10.3. 使用 tracers 强制 Kafka 客户端

检测 Kafka 制作者和消费者客户端，以及 Kafka Streams API 应用程序以实现分布式追踪。

10.3.1. 强制生产者和消费者进行追踪
复制链接

使用 Decorator 模式或 Interceptors 来检测 Java 制作者和使用者应用代码进行追踪。

流程

在每个生产者和消费者应用程序的应用程序代码中：

将 OpenTracing 的 Maven 依赖项添加到制作者或消费者的 pom.xml 文件中。

<dependency>
    <groupId>io.opentracing.contrib</groupId>
    <artifactId>opentracing-kafka-client</artifactId>
    <version>0.1.15.redhat-00001</version>
</dependency>

<dependency>
    <groupId>io.opentracing.contrib</groupId>
    <artifactId>opentracing-kafka-client</artifactId>
    <version>0.1.15.redhat-00001</version>
</dependency>

Copy to Clipboard

Toggle word wrap

使用 Decorator 模式或 Interceptors 检测您的客户端应用程序代码。

使用 Decorator 模式：

// Create an instance of the KafkaProducer:
KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps);

// Create an instance of the TracingKafkaProducer:
TracingKafkaProducer<Integer, String> tracingProducer = new TracingKafkaProducer<>(producer,
        tracer);

// Send:
tracingProducer.send(...);

// Create an instance of the KafkaConsumer:
KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);

// Create an instance of the TracingKafkaConsumer:
TracingKafkaConsumer<Integer, String> tracingConsumer = new TracingKafkaConsumer<>(consumer,
        tracer);

// Subscribe:
tracingConsumer.subscribe(Collections.singletonList("messages"));

// Get messages:
ConsumerRecords<Integer, String> records = tracingConsumer.poll(1000);

// Retrieve SpanContext from polled record (consumer side):
ConsumerRecord<Integer, String> record = ...
SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer);

// Create an instance of the KafkaProducer:
KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps);

// Create an instance of the TracingKafkaProducer:
TracingKafkaProducer<Integer, String> tracingProducer = new TracingKafkaProducer<>(producer,
        tracer);

// Send:
tracingProducer.send(...);

// Create an instance of the KafkaConsumer:
KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);

// Create an instance of the TracingKafkaConsumer:
TracingKafkaConsumer<Integer, String> tracingConsumer = new TracingKafkaConsumer<>(consumer,
        tracer);

// Subscribe:
tracingConsumer.subscribe(Collections.singletonList("messages"));

// Get messages:
ConsumerRecords<Integer, String> records = tracingConsumer.poll(1000);

// Retrieve SpanContext from polled record (consumer side):
ConsumerRecord<Integer, String> record = ...
SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer);

Copy to Clipboard

Toggle word wrap

使用 Interceptors:

// Register the tracer with GlobalTracer:
GlobalTracer.register(tracer);

// Add the TracingProducerInterceptor to the sender properties:
senderProps.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
          TracingProducerInterceptor.class.getName());

// Create an instance of the KafkaProducer:
KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps);

// Send:
producer.send(...);

// Add the TracingConsumerInterceptor to the consumer properties:
consumerProps.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
          TracingConsumerInterceptor.class.getName());

// Create an instance of the KafkaConsumer:
KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);

// Subscribe:
consumer.subscribe(Collections.singletonList("messages"));

// Get messages:
ConsumerRecords<Integer, String> records = consumer.poll(1000);

// Retrieve the SpanContext from a polled message (consumer side):
ConsumerRecord<Integer, String> record = ...
SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer);

// Register the tracer with GlobalTracer:
GlobalTracer.register(tracer);

// Add the TracingProducerInterceptor to the sender properties:
senderProps.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
          TracingProducerInterceptor.class.getName());

// Create an instance of the KafkaProducer:
KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps);

// Send:
producer.send(...);

// Add the TracingConsumerInterceptor to the consumer properties:
consumerProps.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
          TracingConsumerInterceptor.class.getName());

// Create an instance of the KafkaConsumer:
KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);

// Subscribe:
consumer.subscribe(Collections.singletonList("messages"));

// Get messages:
ConsumerRecords<Integer, String> records = consumer.poll(1000);

// Retrieve the SpanContext from a polled message (consumer side):
ConsumerRecord<Integer, String> record = ...
SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer);

Copy to Clipboard

Toggle word wrap

10.3.1.1. Decorator 模式中的自定义范围名称
复制链接

span 是 Jaeger 中的逻辑工作单元，具有操作名称、开始时间和持续时间。

要使用 Decorator 模式来检测您的制作者和消费者应用程序，请在创建 TracingKafkaProducer 和 TracingKafka Consumer 对象时传递 BiFunction 对象作为额外参数来定义自定义 span 名称。OpenTracing Apache Kafka Client Instrumentation 库包含多个内置范围名称。

示例：使用自定义 span 名称在 Decorator 模式中检测客户端应用程序代码

// Create a BiFunction for the KafkaProducer that operates on (String operationName, ProducerRecord consumerRecord) and returns a String to be used as the name:

BiFunction<String, ProducerRecord, String> producerSpanNameProvider =
    (operationName, producerRecord) -> "CUSTOM_PRODUCER_NAME";

// Create an instance of the KafkaProducer:
KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps);

// Create an instance of the TracingKafkaProducer
TracingKafkaProducer<Integer, String> tracingProducer = new TracingKafkaProducer<>(producer,
        tracer,
        producerSpanNameProvider);

// Spans created by the tracingProducer will now have "CUSTOM_PRODUCER_NAME" as the span name.

// Create a BiFunction for the KafkaConsumer that operates on (String operationName, ConsumerRecord consumerRecord) and returns a String to be used as the name:

BiFunction<String, ConsumerRecord, String> consumerSpanNameProvider =
    (operationName, consumerRecord) -> operationName.toUpperCase();

// Create an instance of the KafkaConsumer:
KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);

// Create an instance of the TracingKafkaConsumer, passing in the consumerSpanNameProvider BiFunction:

TracingKafkaConsumer<Integer, String> tracingConsumer = new TracingKafkaConsumer<>(consumer,
        tracer,
        consumerSpanNameProvider);

// Spans created by the tracingConsumer will have the operation name as the span name, in upper-case.
// "receive" -> "RECEIVE"

// Create a BiFunction for the KafkaProducer that operates on (String operationName, ProducerRecord consumerRecord) and returns a String to be used as the name:

BiFunction<String, ProducerRecord, String> producerSpanNameProvider =
    (operationName, producerRecord) -> "CUSTOM_PRODUCER_NAME";

// Create an instance of the KafkaProducer:
KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps);

// Create an instance of the TracingKafkaProducer
TracingKafkaProducer<Integer, String> tracingProducer = new TracingKafkaProducer<>(producer,
        tracer,
        producerSpanNameProvider);

// Spans created by the tracingProducer will now have "CUSTOM_PRODUCER_NAME" as the span name.

// Create a BiFunction for the KafkaConsumer that operates on (String operationName, ConsumerRecord consumerRecord) and returns a String to be used as the name:

BiFunction<String, ConsumerRecord, String> consumerSpanNameProvider =
    (operationName, consumerRecord) -> operationName.toUpperCase();

// Create an instance of the KafkaConsumer:
KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);

// Create an instance of the TracingKafkaConsumer, passing in the consumerSpanNameProvider BiFunction:

TracingKafkaConsumer<Integer, String> tracingConsumer = new TracingKafkaConsumer<>(consumer,
        tracer,
        consumerSpanNameProvider);

// Spans created by the tracingConsumer will have the operation name as the span name, in upper-case.
// "receive" -> "RECEIVE"

Copy to Clipboard

Toggle word wrap

10.3.1.2. 内置范围名称
复制链接

在定义自定义 span 名称时，您可以在 ClientSpanNameProvider 类中使用以下 BiFunctions。如果没有指定 spanNameProvider，则使用 CONSUMER_OPERATION_NAME 和 PRODUCER_OPERATION_NAME。

Expand

BiFunction	描述
`CONSUMER_OPERATION_NAME, PRODUCER_OPERATION_NAME`	将 `operationName` 返回为 span name: "receive" for使用者，并为生产者返回"send"。
`CONSUMER_PREFIXED_OPERATION_NAME（String 前缀）, PRODUCER_PREFIXED_OPERATION_NAME（String 前缀）`	返回 `prefix` 和 `operationName 的 String 连接`。
`CONSUMER_TOPIC, PRODUCER_TOPIC`	返回消息发送到或检索到的主题的名称，格式为 `（record.topic（））`。
`PREFIXED_CONSUMER_TOPIC（String 前缀）, PREFIXED_PRODUCER_TOPIC（String 前缀）`	返回 `前缀` 的字符串串联和主题名称，格式为 `（record.topic（））`。
`CONSUMER_OPERATION_NAME_TOPIC, PRODUCER_OPERATION_NAME_TOPIC`	返回操作名称和主题名称 `："operationName - record.topic（）"。`
`CONSUMER_PREFIXED_OPERATION_NAME_TOPIC（String 前缀）、PRODUCER_PREFIXED_OPERATION_NAME_TOPIC（String 前缀）`	返回 `前缀` 的字符串串联和 `"operationName - record.topic（）"。`

返回顶部

10.3. 使用 tracers 强制 Kafka 客户端

10.3.1. 强制生产者和消费者进行追踪
复制链接

10.3.1.1. Decorator 模式中的自定义范围名称
复制链接

10.3.1.2. 内置范围名称
复制链接

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links

10.3. 使用 tracers 强制 Kafka 客户端

10.3.1. 强制生产者和消费者进行追踪复制链接链接已复制到粘贴板!

10.3.1.1. Decorator 模式中的自定义范围名称复制链接链接已复制到粘贴板!

10.3.1.2. 内置范围名称复制链接链接已复制到粘贴板!

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links

10.3.1. 强制生产者和消费者进行追踪
复制链接

10.3.1.1. Decorator 模式中的自定义范围名称
复制链接

10.3.1.2. 内置范围名称
复制链接