Kafka: la rédaction de la coutume de sérialiseur

Je suis en train de construire un POC avec Kafka 0.8.1. Je suis en utilisant ma propre classe java comme Kafka message qui a un tas de types de données Chaîne. Je ne peux pas utiliser la valeur par défaut sérialiseur de la classe ou de la Chaîne sérialiseur de classe qui vient avec Kafka de la bibliothèque. Je suppose que j'ai besoin d'écrire mon propre sérialiseur et le nourrir pour le producteur propriétés.
Si vous êtes conscient de l'écriture d'un exemple personnalisé sérialiseur Kafka (en java), merci de le faire partager.
Apprécie beaucoup, merci beaucoup.

  • En fait il est de coutume de Codeur pour kafkaMessage
  • Est-il possible d'exécuter un Kafka application sur i7 ?
InformationsquelleAutor Paaji | 2014-05-20
  1. 41

    Les choses nécessaires à l'écriture d'une coutume sérialiseur sont:

    1. Mettre en œuvre Encoder avec un objet spécifié pour le générique
      • La fourniture d'une VerifiableProperties constructeur est tenu
    2. Remplacer toBytes(...) méthode en s'assurant qu'un tableau d'octets est retourné
    3. Injecter le sérialiseur classe en ProducerConfig

    Déclarant une coutume sérialiseur pour un producteur

    Comme vous l'avez mentionné dans votre question, Kafka fournit un moyen de déclarer un spécifique sérialiseur pour un producteur. Le sérialiseur classe est définie dans un ProducerConfig instance et que cette instance est utilisé pour construire l'souhaité Producer classe.

    Si vous suivez Kafka Producteur Exemple vous permettra de construire des ProducerConfig via un Properties objet. Lors de la construction de votre fichier de propriétés, assurez-vous d'inclure:

    props.put("serializer.class", "path.to.your.CustomSerializer");

    Avec le chemin d'accès à la classe que vous souhaitez Kafka à utiliser pour sérialiser les messages avant de les ajoutant dans le journal.

    Créer un sérialiseur que Kafka comprend

    Écrit une coutume sérialiseur que Kafka pouvoir interpréter correctement nécessite la mise en œuvre de la Encoder[T] scala classe que Kafka fournit. La mise en œuvre des traits dans le langage java est bizarre, mais la méthode suivante travaillé pour la sérialisation JSON dans mon projet:

    public class JsonEncoder implements Encoder<Object> {
    private static final Logger logger = Logger.getLogger(JsonEncoder.class);
    //instantiating ObjectMapper is expensive. In real life, prefer injecting the value.
    private static final ObjectMapper objectMapper = new ObjectMapper();

    public JsonEncoder(VerifiableProperties verifiableProperties) {
        /* This constructor must be present for successful compile. */
    }

    @Override
    public byte[] toBytes(Object object) {
        try {
            return objectMapper.writeValueAsString(object).getBytes();
        } catch (JsonProcessingException e) {
            logger.error(String.format("Json processing failed for object: %s", object.getClass().getName()), e);
        }
        return "".getBytes();
    }
}

    Votre question nous donne l'impression que vous êtes en utilisant un objet (appelons cela CustomMessage) pour tous les messages ajoutés à votre journal. Si c'est le cas, votre sérialiseur pourrait ressembler à ceci:

    package com.project.serializer;

public class CustomMessageEncoder implements Encoder<CustomMessage> {
    public CustomMessageEncoder(VerifiableProperties verifiableProperties) {
        /* This constructor must be present for successful compile. */
    }

    @Override
    public byte[] toBytes(CustomMessage customMessage) {
        return customMessage.toBytes();
    }
}

    Qui risquerait de laisser votre propriété config ressemble à ceci:

    props.put("serializer.class", "path.to.your.CustomSerializer");
    • Merci pour ce Sam B. Extrêmement utile.
    • Comment avez-vous désérialiser le tableau d'octets dans kafka consommateur maintenant que nous avons sérialisé l'objet?
    • Une suggestion: puisque la construction de ObjectMapper est un poids lourd de l'opération, et depuis ils sont thread-safe, après la construction, il est logique de créer un statique de la dernière instance de pour codeur/décodeur. Sinon, la construction prendra 10x plus que le réel de la lecture/écriture.
    • bon point. Je suis sûr que pour la plupart des applications, la meilleure approche serait d'injecter le ObjectMapper, mais comme vous l'avez dit, pour cet exemple montre qu'une seule instanciation est une amélioration.
    • Oui, l'injection d'un correctement configuré instance de sens que dans de nombreux cas. Je voulais juste dire cela car les performances de l'effet est significatif, et parce que le code est souvent la cut-n-collé verbatim
    • si cette méthode fonctionne avec la version 0.8.2 de kafka producteur?
    • Je ne pense pas que vous en avez besoin pour 0.8.2 parce que Kafka 0.8.2 a son propre Sérialiseur et Deserializer interface.

    InformationsquelleAutor Sam Berry
  2. 12

    Vous avez besoin pour mettre en œuvre la fois d'encoder et de decoder

    public class JsonEncoder implements Encoder<Object> {
        private static final Logger LOGGER = Logger.getLogger(JsonEncoder.class);

        public JsonEncoder(VerifiableProperties verifiableProperties) {
            /* This constructor must be present for successful compile. */
        }

        @Override
        public byte[] toBytes(Object object) {
            ObjectMapper objectMapper = new ObjectMapper();
            try {
                return objectMapper.writeValueAsString(object).getBytes();
            } catch (JsonProcessingException e) {
                LOGGER.error(String.format("Json processing failed for object: %s", object.getClass().getName()), e);
            }
            return "".getBytes();
        }
    }

    Le décodeur de code

    public class JsonDecoder  implements Decoder<Object> {
    private static final Logger LOGGER = Logger.getLogger(JsonEncoder.class);
    public JsonDecoder(VerifiableProperties verifiableProperties) {
        /* This constructor must be present for successful compile. */
    }

    @Override
    public Object fromBytes(byte[] bytes) {
        ObjectMapper objectMapper = new ObjectMapper();
        try {
            return objectMapper.readValue(bytes, Map.class);
        } catch (IOException e) {
            LOGGER.error(String.format("Json processing failed for object: %s", bytes.toString()), e);
        }
        return null;
    }
}

    La pom entrée

    <dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.4.1.3</version>
</dependency>

    Définir la valeur par défaut du codeur dans le Kafka de la propriété

    properties.put("serializer.class","kafka.serializer.DefaultEncoder");

    L'écrivain et le lecteur de code est comme suit

    byte[] bytes = encoder.toBytes(map);
        KeyedMessage<String, byte[]> message =new KeyedMessage<String, byte[]>(this.topic, bytes);

JsonDecoder decoder = new JsonDecoder(null);
Map map = (Map) decoder.fromBytes(it.next().message());
    • Comment créer un exemple de codeur? VerifiableProperties peut être nulle?
    InformationsquelleAutor Harvinder Singh