senden Benutzerdefinierte Java-Objekte zu Kafka Thema

Question

ich mein benutzerdefinierten Java-Objekt und will in gebaut Serialisierung nutzen JVM es zu einem Kafka Thema zu senden, aber die Serialisierung nicht mit folgenden Fehlern

org.apache.kafka. common.errors.SerializationException: Kann nicht Wert der Klasse com.spring.kafka.Payload Klasse org.apache.kafka.common.serialization.ByteArraySerializer angegeben in value.serializer

Payload konvertieren. Java

public class Payload implements Serializable { 

    private static final long serialVersionUID = 123L; 

    private String name="vinod"; 

    private int anInt = 5; 

    private Double aDouble = new Double("5.0"); 

    public String getName() { 
     return name; 
    } 

    public void setName(String name) { 
     this.name = name; 
    } 

    public int getAnInt() { 
     return anInt; 
    } 

    public void setAnInt(int anInt) { 
     this.anInt = anInt; 
    } 

    public Double getaDouble() { 
     return aDouble; 
    } 

    public void setaDouble(Double aDouble) { 
     this.aDouble = aDouble; 
    } 

} 

Während meiner Bildung von Erzeuger habe ich die folgenden Eigenschaften festgelegt

<entry key="key.serializer" 
         value="org.apache.kafka.common.serialization.ByteArraySerializer" /> 
       <entry key="value.serializer" 
         value="org.apache.kafka.common.serialization.ByteArraySerializer" /> 

Meine Sende invoke wie unten

kafkaProducer.send(new ProducerRecord<String, Payload>("test", new Payload())); 

ist Was richtige Weg ist, ein eigenes Java-Objekt senden durch einen Produzenten zu einem Kafka-Thema?

Answer 1

Wir haben 2 Optionen, wie nachstehend aufgeführt

1) Wenn wir beabsichtigen, benutzerdefinierte Java-Objekte Hersteller zu schicken, brauchen wir einen Serializer zu schaffen, die org.apache.kafka.common.serialization.Serializer implementiert und passieren, dass Serializer Klasse bei der Erstellung des Produzenten

Code Referenz unter

public class PayloadSerializer implements org.apache.kafka.common.serialization.Serializer { 

    public void configure(Map map, boolean b) { 

    } 

    public byte[] serialize(String s, Object o) { 

     try { 
      ByteArrayOutputStream baos = new ByteArrayOutputStream(); 
      ObjectOutputStream oos = new ObjectOutputStream(baos); 
      oos.writeObject(o); 
      oos.close(); 
      byte[] b = baos.toByteArray(); 
      return b; 
     } catch (IOException e) { 
      return new byte[0]; 
     } 
    } 

    public void close() { 

    } 
} 

und den Wert Serializer entsprechend eingestellt

<entry key="value.serializer" 
         value="com.spring.kafka.PayloadSerializer" /> 

2) Keine Notwendigkeit, benutzerdefinierte Serializer-Klasse zu erstellen. Verwenden Sie die vorhandenen ByteArraySerializer, aber beim Senden folgen Sie den Anweisungen

Java Object -> String (JSON represenation Vorzugsweise statt toString) -> byteArray

Answer 2

Da Sie ByteArraySerializer verwenden, müssen Sie ein Byte [] Producer instanziieren.

Producer<byte[],byte[]> producer = new KafkaProducer<>(props); 

und dann, während das Byte produzieren passieren [] nach der Serialisierung oder ein anderes Verfahren, zum Beispiel

producer.send(new ProducerRecord<byte[],byte[]>("test", new Payload().toString().getBytes())); 

Wenn Sie nur ein Payload-Objekt an den Erzeuger sind vorbei, dann wird es besser sein, Haben Schlüssel Serializer und Wert Serializer als was auch immer Sie vorhaben zu übergeben und während des Lesens müssen Sie aus diesen Daten lesen.

Es empfiehlt sich, Serializable und ByteArraySerializer/ByteArrayDeserializer zu verwenden.