Kafka 消息序列化和反序列化（下）

有序列化就會有反序列化，反序列化的操作是在Kafka Consumer中完成的，使用起來只需要配置一下key.deserializer和value.deseriaizer。對應(yīng)上面自定義的Company類型的Deserializer就需要實現(xiàn)org.apache.kafka.common.serialization.Deserializer接口，這個接口同樣有三個方法：

public void configure(Map<String, ?> configs, boolean isKey)：用來配置當(dāng)前類。
public byte[] serialize(String topic, T data)：用來執(zhí)行反序列化。如果data為null建議處理的時候直接返回null而不是拋出一個異常。
public void close()：用來關(guān)閉當(dāng)前序列化器。
下面就來看一下DemoSerializer對應(yīng)的反序列化的DemoDeserializer，詳細(xì)代碼如下：

public class DemoDeserializer implements Deserializer<Company> {
    public void configure(Map<String, ?> configs, boolean isKey) {}
    public Company deserialize(String topic, byte[] data) {
        if (data == null) {
            return null;
        }
        if (data.length < 8) {
            throw new SerializationException("Size of data received by DemoDeserializer is shorter than expected!");
        }
        ByteBuffer buffer = ByteBuffer.wrap(data);
        int nameLen, addressLen;
        String name, address;
        nameLen = buffer.getInt();
        byte[] nameBytes = new byte[nameLen];
        buffer.get(nameBytes);
        addressLen = buffer.getInt();
        byte[] addressBytes = new byte[addressLen];
        buffer.get(addressLen);
        try {
            name = new String(nameBytes, "UTF-8");
            address = new String(addressBytes, "UTF-8");
        } catch (UnsupportedEncodingException e) {
            throw new SerializationException("Error occur when deserializing!");
        }
        return new Company(name,address);
    }
    public void close() {}
}

有些讀者可能對新版的Consumer不是很熟悉，這里順帶著舉一個完整的消費示例，并以DemoDeserializer作為消息Value的反序列化器。

Properties properties = new Properties();
properties.put("bootstrap.servers", brokerList);
properties.put("group.id", consumerGroup);
properties.put("session.timeout.ms", 10000);
properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.put("value.deserializer", "com.hidden.client.DemoDeserializer");
properties.put("client.id", "hidden-consumer-client-id-zzh-2");
KafkaConsumer<String, Company> consumer = new KafkaConsumer<String, Company>(properties);
consumer.subscribe(Arrays.asList(topic));
try {
    while (true) {
        ConsumerRecords<String, Company> records = consumer.poll(100);
        for (ConsumerRecord<String, Company> record : records) {
            String info = String.format("topic=%s, partition=%s, offset=%d, consumer=%s, country=%s",
                    record.topic(), record.partition(), record.offset(), record.key(), record.value());
            System.out.println(info);
        }
        consumer.commitAsync(new OffsetCommitCallback() {
            public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                if (exception != null) {
                    String error = String.format("Commit failed for offsets {}", offsets, exception);
                    System.out.println(error);
                }
            }
        });
    }
} finally {
    consumer.close();
}

有些時候自定義的類型還可以和Avro、ProtoBuf等聯(lián)合使用，而且這樣更加的方便快捷，比如我們將前面Company的Serializer和Deserializer用Protostuff包裝一下，由于篇幅限制，筆者這里只羅列出對應(yīng)的serialize和deserialize方法，詳細(xì)參考如下：

public byte[] serialize(String topic, Company data) {
    if (data == null) {
        return null;
    }
    Schema schema = (Schema) RuntimeSchema.getSchema(data.getClass());
    LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
    byte[] protostuff = null;
    try {
        protostuff = ProtostuffIOUtil.toByteArray(data, schema, buffer);
    } catch (Exception e) {
        throw new IllegalStateException(e.getMessage(), e);
    } finally {
        buffer.clear();
    }
    return protostuff;
}

public Company deserialize(String topic, byte[] data) {
    if (data == null) {
        return null;
    }
    Schema schema = RuntimeSchema.getSchema(Company.class);
    Company ans = new Company();
    ProtostuffIOUtil.mergeFrom(data, ans, schema);
    return ans;
}

如果Company的字段很多，我們使用Protostuff進(jìn)一步封裝一下的方式就顯得簡潔很多。不過這個不是最主要的，而最主要的是經(jīng)過Protostuff包裝之后，這個Serializer和Deserializer可以向前兼容（新加字段采用默認(rèn)值）和向后兼容（忽略新加字段），這個特性Avro和Protobuf也都具備。

自定義的類型有一個不得不面對的問題就是Kafka Producer和Kafka Consumer之間的序列化和反序列化的兼容性，試想對于StringSerializer來說，Kafka Consumer可以順其自然的采用StringDeserializer，不過對于Company這種專用類型，某個服務(wù)使用DemoSerializer進(jìn)行了序列化之后，那么下游的消費者服務(wù)必須也要實現(xiàn)對應(yīng)的DemoDeserializer。再者，如果上游的Company類型改變，下游也需要跟著重新實現(xiàn)一個新的DemoSerializer，這個后面所面臨的難題可想而知。所以，如無特殊需要，筆者不建議使用自定義的序列化和反序列化器；如有業(yè)務(wù)需要，也要使用通用的Avro、Protobuf、Protostuff等序列化工具包裝，盡可能的實現(xiàn)得更加通用且向前后兼容。

題外話，對于Kafka的“深耕者”Confluent來說，還有其自身的一套序列化和反序列化解決方案（io.confluent.kafka.serializer.KafkaAvroSerializer），GitHub上有相關(guān)資料，讀者如有興趣可以自行擴展學(xué)習(xí)。

本文的重點是你有沒有收獲與成長，其余的都不重要，希望讀者們能謹(jǐn)記這一點。同時我經(jīng)過多年的收藏目前也算收集到了一套完整的學(xué)習(xí)資料，包括但不限于：分布式架構(gòu)、高可擴展、高性能、高并發(fā)、Jvm性能調(diào)優(yōu)、Spring，MyBatis，Nginx源碼分析，Redis，ActiveMQ、、Mycat、Netty、Kafka、Mysql、Zookeeper、Tomcat、Docker、Dubbo、Nginx等多個知識點高級進(jìn)階干貨，希望對想成為架構(gòu)師的朋友有一定的參考和幫助

需要更詳細(xì)思維導(dǎo)圖和以下資料的可以加一下技術(shù)交流分享群：“708 701 457”免費獲取