如何处理从 Confluent Kafka 消费者收到的结果？答案

【问题标题】：How to handle the result received from Confluent Kafka consumer?如何处理从 Confluent Kafka 消费者收到的结果？
【发布时间】：2019-11-22 18:11:08
【问题描述】：

您好，我正在开发 Confluent Kafka。我有一个返回 ConsumeResult 的消费者。下面是我的消费者实现。

public ConsumeResult<string, GenericRecord> Consume(string topic)
    {
      consumer.Subscribe(topic);
      ConsumeResult<string, GenericRecord> result;
      try
      {
        result = consumer.Consume();
        return result;
      }
      catch (Exception e)
      {
        this.logger.Error("KafkaClient", $"Error sending message '{e.Message}'");
        return null;
      }
    }

我称这个方法为

public ConsumeResult<string, GenericRecord> SubscribeAsync()
    {
      return this.consumerClient.Consume(productEventTopicName);
    }

最后我从控制器调用

var productEvents = this.eventDispatcher.SubscribeAsync();

现在我想处理这个结果。我知道我可以得到一个字段

response.Message.Value.TryGetValue("Version", out object result);

我想一口气获得所有领域。我的困惑是我是否需要根据从消费者那里收到的数据或我可以使用的现有模型来选择所需的模型？如果我使用现有模型，那么如果明天新字段出现在消费者结果中，那么我们的模型将如何处理它？有人可以帮助我理解这一点吗？任何帮助，将不胜感激。谢谢。

【问题讨论】：

标签： .net apache-kafka avro confluent-schema-registry confluent-platform

【解决方案1】：

当您使用 GenericRecord 时，反序列化器将联系模式注册表以在生产者发送时提取新的模式信息。

但是，为了解析新字段，您必须停止消费者并显式添加代码以相应地处理这些属性。这个问题不是卡夫卡特有的

【讨论】：