Что означает десериализация ключа в классе Consumer в Confluent Kafka? - PullRequest
Новая
       42

Что означает десериализация ключа в классе Consumer в Confluent Kafka?

0 голосов
/ 27 апреля 2018

В соответствующей документации kafka сказано, что класс Consumer определен следующим образом: 

Class Consumer<TKey, TValue>

Класс потребителя, описанный выше, реализует Apache Kafka высокого уровня. потребитель (с десериализацией ключа и значения).

Я понимаю, что TKey и TValue предназначены для десериализации ключа, отправленного производителем. Например, что-то вроде

Отправка в ключ от производителя будет выглядеть как 

var deliveryReport = producer.ProduceAsync(topicName, key, val);

Получение строкового ключа на стороне потребителя будет выглядеть как 

using (var consumer = new Consumer<Ignore, string>(constructConfig(brokerList, false), null, new StringDeserializer(Encoding.UTF8)))
{
    consumer.Subscribe(topics);

    Console.WriteLine($"Started consumer, Ctrl-C to stop consuming");

    var cancelled = false;
    Console.CancelKeyPress += (_, e) => {
        e.Cancel = true; // prevent the process from terminating.
        cancelled = true;
    };

    while (!cancelled)
    {
        Message<Ignore, string> msg;
        if (!consumer.Consume(out msg, TimeSpan.FromMilliseconds(100)))
        {
            continue;
        }

        Console.WriteLine($"Topic: {msg.Topic} Partition: {msg.Partition} Offset: {msg.Offset} {msg.Value}");
    }
}

Поскольку мы передаем ключ, Потребитель инициализируется как 

Consumer<Ignore, string>

и сообщение инициализируется как 

Message<Ignore, String>

После всего этого, мой вопрос: что на самом деле означает десериализация ключа ? И зачем нам это делать? Кроме того, зачем нам нужно передавать пару ключ-значение Ignore, String для выполнения десериализации?

1 Ответ

0 голосов
/ 28 апреля 2018

зачем нам нужно передавать пару ключ-значение Ignore, String для выполнения десериализации?

Вам не нужно передавать эти конкретные настройки. Вам необходимо соответствовать настройкам производителя. Или, если вы не уверены, вы бы дали объект байтового массива как для ключа, так и для значения.

Если производитель не отправил ключ, например, ноль, десериализовать нечего. Я полагаю, для этого предназначен класс Ignore. Обратите внимание, что вы не указали ключевой класс Deserializer, но указали значение 

null, new StringDeserializer(Encoding.UTF8))

Все сообщения Kafka содержат пары ключей и значений только в байтах. Производители используют сериализаторы, и как потребитель вы должны десериализовать. В идеале вы десериализуете сообщения в реальные объекты, такие как строки или объекты JSON или Avro, Protobuf и т. Д. И т. Д.

По умолчанию ключи определяют, из каких разделов темы исходят сообщения, из которых вы будете получать сообщения. Нулевой ключ будет равномерно распределен по теме. В противном случае приложение-производитель может определить свои собственные разделители и отправлять данные везде, где их логика решает

...