Как отправить пользовательский объект в тему Кафки с продюсером - PullRequest
Новая
       6

Как отправить пользовательский объект в тему Кафки с продюсером

0 голосов
/ 21 сентября 2018

Я хотел бы отправить свой класс Account с Producer в мою тему Kafka, затем я объединюсь с Kafka Stream.Тем не менее, я не могу отправить объект Я получаю сообщение об ошибке:

Причина: org.apache.kafka.common.KafkaException: bank.Account не является экземпляром org.apache.kafka.common.serialization.Serializer

Мой класс продюсера: 

 public static void main(String[] args) {

        DataAccess dataAccess = new DataAccess();
        List<Account> accountList = dataAccess.read();

        final Logger logger = LoggerFactory.getLogger(Producer.class);
        Properties properties = new Properties();

        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,LongSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,Account.class.getName());


        KafkaProducer<Long,Account> producer = new KafkaProducer<Long, Account>(properties);



        for (Account account : accountList) {

            ProducerRecord<Long,Account> record = new ProducerRecord<Long, Account>("bank_account",account.getFromId(),account);


            producer.send(record, new Callback() {
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e == null) {
                        logger.info("Record sent successfully. \n "+ "Topic : "+recordMetadata.topic() +"\n"+
                                "Partition : " + recordMetadata.partition() + "\n"+
                                "Offset : " +recordMetadata.offset() +"\n"+
                                "Timestamp: " +recordMetadata.timestamp() +"\n");
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e1) {
                            e1.printStackTrace();
                        }

                    }
                    else{
                        logger.info("Error sending producer");
                    }
                }
            });
        }


        producer.flush();
        producer.close();
    }

Выдает ошибку в этой строке: 

KafkaProducer<Long,Account> producer = new KafkaProducer<Long, Account>(properties);

Класс моего аккаунта: 

public class Account {

    private long fromId;
    private long amount;
    private long toId;
    private ZonedDateTime time;
}

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

1 Ответ

0 голосов
/ 21 сентября 2018

Эта строка

properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,Account.class.getName());

Вы должны реализовать свой собственный Serializer класс .Это не может быть простой класс.

Некоторые люди используют JSON для сериализации, другие используют Avro или Protobuf.Однако вы получаете данные в byte[] только детали реализации.

...