Confluent.Kafka.Extensions.Diagnostics 0.5.0

Confluent.Kafka.Extensions.Diagnostics

Confluent.Kafka.Extensions.Diagnostics 包通过 活动 API 为 Confluent.Kafka 库启用仪器化。

安装

Install-Package Confluent.Kafka.Extensions.Diagnostics

用法

生产者

生产者仪器化是通过包装类完成的，因此不需要重写生产者用法。然而，为了启用生产者仪器化，应在与 Build 代替调用 BuildWithInstrumentation 方法。之后，将仪器化所有生产调用（同步和异步）。

using Confluent.Kafka;
using Confluent.Kafka.Extensions.Diagnostics;


using var producer =
    new ProducerBuilder<Null, string>(new ProducerConfig(new ClientConfig { BootstrapServers = "localhost:9092" }))
        .SetKeySerializer(Serializers.Null)
        .SetValueSerializer(Serializers.Utf8)
        .BuildWithInstrumentation();

await producer.ProduceAsync("topic", new Message<Null, string> { Value = "Hello World!" });

消费者

不幸的是，Confluent.Kafka 库的消费者接口并不非常灵活。因此，通过消费者本身的扩展方法实现仪器化。因此，应将消费者用法重写如下

using Confluent.Kafka;
using Confluent.Kafka.Extensions.Diagnostics;

using var consumer = new ConsumerBuilder<Ignore, string>(
        new ConsumerConfig(new ClientConfig { BootstrapServers = "localhost:9092" })
        {
            GroupId = "group", AutoOffsetReset = AutoOffsetReset.Earliest
        })
    .SetValueDeserializer(Deserializers.Utf8)
    .Build();

consumer.Subscribe("topic");

try
{
    while (true)
    {
        try
        {
            consumer.ConsumeWithInstrumentation((result) =>
            {
                Console.WriteLine(result.Message.Value);
            }, 2000);
        }
        catch (ConsumeException e)
        {
            Console.WriteLine($"Error occured: {e.Error.Reason}");
        }
    }
}
catch (OperationCanceledException)
{
    consumer.Close();
}