如何在融合的 kafka C# 中读取 GenericRecord 特定数据答案

【问题标题】：How to read GenericRecord specific data in confluent kafka C#如何在融合的 kafka C# 中读取 GenericRecord 特定数据
【发布时间】：2019-09-12 17:01:39
【问题描述】：

这是我尝试读取 Avro 通用记录表单消费者的简单代码 sn-p：

using (var schemaRegistry = new CachedSchemaRegistryClient(new SchemaRegistryConfig { SchemaRegistryUrl = schemaRegistryUrl }))
  using (var consumer = new 
        ConsumerBuilder<string, GenericRecord>(new ConsumerConfig { BootstrapServers = bootstrapServers, GroupId = groupName })
                        .SetKeyDeserializer(new AsyncAvroDeserializer<string>(schemaRegistry).AsSyncOverAsync())
                        .SetValueDeserializer(new AsyncAvroDeserializer<GenericRecord>(schemaRegistry).AsSyncOverAsync())
                        .SetErrorHandler((_, e) => Console.WriteLine($"Error: {e.Reason}"))
                        .Build())
                {
      consumer.Subscribe(topicName);

      try
       {
           while (true)
             {
                try
                 {
                    var consumeResult = consumer.Consume();
                    Console.WriteLine($"Key: {consumeResult.Message.Key}\nValue: {consumeResult.Value}");
                    Console.WriteLine(consumeResult.Value.Schema);
                            Console.WriteLine(consumeResult.Value.Schema["favorite_number"]);

                  }
                 catch (ConsumeException e)
                   {
                       Console.WriteLine($"Consume error: {e.Error.Reason}");
                            }
                        }
                    }
                    catch (OperationCanceledException)
                    {
                        // commit final offsets and leave the group.
                        consumer.Close();
                    }
                }

如您所见，我可以记录架构但不知道如何获取其数据值。

 Console.WriteLine(consumeResult.Value.Schema);

 Console.WriteLine(consumeResult.Value.Schema["favorite_number"]);

虽然这个对象consumeResult.Value中的架构和数据是这样的：

{Schema: {"type":"record","name":"User","namespace":"Confluent.Kafka.Examples.AvroSpecific","fields":[{"name":"name" ,"type":"string"},{"name":"favorite_number","type":["int","null"]},{"name":"favorite_color","type":["string ","null"]}]}, contents: { name: sfs, favorite_number: 41, favorite_color: blue, }}

我想读取内容数据。

【问题讨论】：

consumeResult.Value 是字符串吗？或者那是一个动态对象？ consumeResult.Value.contents["favorite_number"] 不起作用？否则它看起来需要先进行 JSON 反序列化。
value是GenericRecord的TValue，而consumeResult.Value.contents["favorite_number"]，内容不可访问。

标签： c# apache-kafka avro confluent-schema-registry confluent-platform

【解决方案1】：

content 不可访问

我不认为Value.contents 是你想要的。特别是因为该属性是private，正如您所提到的

getter 的定义类似于字典 - Source code

试试consumeResult.Value["favorite_number"]

当您执行consumeResult.Value.Schema["favorite_number"] 时，您会在架构中获得Field 对象，而不是外部记录中的字段值。 - Source code

【讨论】：