Kafka Streams: сериализация / десериализация POJO - PullRequest
Новая
       8

Kafka Streams: сериализация / десериализация POJO

0 голосов
/ 16 мая 2018

Какой класс / метод в Kafka Streams мы можем использовать для сериализации / десериализации Java-объекта в байтовый массив ИЛИ наоборот? Следующая ссылка предлагает использовать ByteArrayOutputStream & ObjectOutputStream, но они не являются поточно-ориентированными.

Отправка пользовательских объектов Java в тему Kafka

Существует еще один вариант использования ObjectMapper, ObjectReader (для поточно-ориентированного), но он конвертируется из POJO -> JSON -> bytearray. Кажется, этот вариант является обширным. Хотел проверить, есть ли прямой способ перевести объект в bytearray и наоборот, который является потокобезопасным. Пожалуйста, предложите 

import org.apache.kafka.common.serialization.Serializer;
public class HouseSerializer<T> implements Serializer<T>{
    private Class<T> tClass;
    public HouseSerializer(){

    }

    @SuppressWarnings("unchecked")
    @Override
    public void configure(Map configs, boolean isKey) {
        tClass = (Class<T>) configs.get("POJOClass");       
    }

    @Override
    public void close() {
    }

    @Override
    public byte[] serialize(String topic, T data) {
        //Object serialization to be performed here
        return null;
    }
}


Примечание: версия Kafka - 0.10.1

1 Ответ

0 голосов
/ 16 мая 2018

Хотел проверить, есть ли прямой способ перевести объект в bytearray

Я бы посоветовал вам использовать Avro serialization с реестром Confluent Schema, если это возможно, но не обязательно. JSON - хороший запасной вариант, но он занимает больше места «на проводе», и поэтому альтернативой будет MsgPack .

См. Пример кода Avro здесь

Приведенный выше пример использует avro-maven-plugin для генерации класса LogLine из файла схемы src/main/resources/avro.

В противном случае, вам нужно как сериализовать ваш объект в байтовый массив , например, строка обычно упаковывается как 

[(length of string) (UTF8 encoded bytes)]

В то время как логические значения представляют собой один бит 0 или 1

потокобезопасный

Я понимаю проблему, но вы обычно не делитесь десериализованными данными между потоками. Вы отправляете / читаете / обрабатываете сообщение для каждого независимого.

...