面试题：自定义 Kafka 消息序列化与反序列化器实现复杂业务场景需求

实现步骤

定义数据结构：首先明确复杂数据结构的具体形式，例如可以用Java类来表示嵌套对象。对于动态字段，可以考虑使用Map来存储。对于加密部分，定义合适的加密和解密方法。
实现序列化器：
- 实现org.apache.kafka.common.serialization.Serializer接口。
- 在serialize方法中，先将对象转换为字节数组。对于嵌套对象，可以递归处理。对于动态字段，按照一定格式（如JSON）进行编码。对需要加密的部分，调用加密方法进行加密。
实现反序列化器：
- 实现org.apache.kafka.common.serialization.Deserializer接口。
- 在deserialize方法中，先将字节数组转换回原始数据形式。对加密部分进行解密，然后按照序列化时的格式解析动态字段和嵌套对象。
注册序列化器和反序列化器：
- 在Kafka生产者和消费者配置中，分别指定自定义的序列化器和反序列化器。例如在Java中，生产者配置：

Properties props = new Properties();
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "com.example.CustomSerializer");

消费者配置：

Properties props = new Properties();
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "com.example.CustomDeserializer");

关键代码逻辑

自定义序列化器示例（Java）：

import org.apache.kafka.common.serialization.Serializer;
import java.nio.ByteBuffer;
import java.util.Map;
import javax.crypto.Cipher;
import javax.crypto.spec.SecretKeySpec;

public class CustomSerializer implements Serializer<ComplexData> {

    private static final String ALGORITHM = "AES";
    private static final byte[] keyValue = new byte[]{'T', 'h', 'i','s', 'I','s', 'A', 'S', 'e', 'c', 'r', 'e', 't', 'K', 'e', 'y'};

    @Override
    public void configure(Map<String,?> configs, boolean isKey) {
        // 初始化配置，例如获取加密密钥等
    }

    @Override
    public byte[] serialize(String topic, ComplexData data) {
        if (data == null) {
            return null;
        }
        // 序列化嵌套对象
        byte[] nestedObjectBytes = serializeNestedObject(data.getNestedObject());
        // 序列化动态字段，假设转为JSON字符串再转字节数组
        byte[] dynamicFieldsBytes = data.getDynamicFields().toString().getBytes();
        // 加密部分
        byte[] encryptedPart = encrypt(data.getEncryptedPart());

        int nestedObjectLength = nestedObjectBytes.length;
        int dynamicFieldsLength = dynamicFieldsBytes.length;
        int encryptedPartLength = encryptedPart.length;

        ByteBuffer buffer = ByteBuffer.allocate(4 + nestedObjectLength + 4 + dynamicFieldsLength + 4 + encryptedPartLength);
        buffer.putInt(nestedObjectLength);
        buffer.put(nestedObjectBytes);
        buffer.putInt(dynamicFieldsLength);
        buffer.put(dynamicFieldsBytes);
        buffer.putInt(encryptedPartLength);
        buffer.put(encryptedPart);

        return buffer.array();
    }

    private byte[] serializeNestedObject(NestedObject nestedObject) {
        // 递归序列化嵌套对象
        // 例如，将嵌套对象的各个字段转为字节数组并拼接
        return new byte[0];
    }

    private byte[] encrypt(String value) {
        try {
            SecretKeySpec secretKey = new SecretKeySpec(keyValue, ALGORITHM);
            Cipher cipher = Cipher.getInstance(ALGORITHM);
            cipher.init(Cipher.ENCRYPT_MODE, secretKey);
            return cipher.doFinal(value.getBytes());
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void close() {
        // 清理资源
    }
}

自定义反序列化器示例（Java）：

import org.apache.kafka.common.serialization.Deserializer;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import javax.crypto.Cipher;
import javax.crypto.spec.SecretKeySpec;

public class CustomDeserializer implements Deserializer<ComplexData> {

    private static final String ALGORITHM = "AES";
    private static final byte[] keyValue = new byte[]{'T', 'h', 'i','s', 'I','s', 'A', 'S', 'e', 'c', 'r', 'e', 't', 'K', 'e', 'y'};

    @Override
    public void configure(Map<String,?> configs, boolean isKey) {
        // 初始化配置，例如获取加密密钥等
    }

    @Override
    public ComplexData deserialize(String topic, byte[] data) {
        if (data == null) {
            return null;
        }
        ByteBuffer buffer = ByteBuffer.wrap(data);
        int nestedObjectLength = buffer.getInt();
        byte[] nestedObjectBytes = new byte[nestedObjectLength];
        buffer.get(nestedObjectBytes);
        NestedObject nestedObject = deserializeNestedObject(nestedObjectBytes);

        int dynamicFieldsLength = buffer.getInt();
        byte[] dynamicFieldsBytes = new byte[dynamicFieldsLength];
        buffer.get(dynamicFieldsBytes);
        Map<String, Object> dynamicFields = deserializeDynamicFields(dynamicFieldsBytes);

        int encryptedPartLength = buffer.getInt();
        byte[] encryptedPart = new byte[encryptedPartLength];
        buffer.get(encryptedPart);
        String decryptedPart = decrypt(encryptedPart);

        return new ComplexData(nestedObject, dynamicFields, decryptedPart);
    }

    private NestedObject deserializeNestedObject(byte[] bytes) {
        // 递归反序列化嵌套对象
        // 例如，从字节数组解析出嵌套对象的各个字段
        return new NestedObject();
    }

    private Map<String, Object> deserializeDynamicFields(byte[] bytes) {
        String json = new String(bytes);
        // 假设使用JSON解析动态字段
        Map<String, Object> dynamicFields = new HashMap<>();
        // 实际实现中使用JSON解析库填充dynamicFields
        return dynamicFields;
    }

    private String decrypt(byte[] encryptedValue) {
        try {
            SecretKeySpec secretKey = new SecretKeySpec(keyValue, ALGORITHM);
            Cipher cipher = Cipher.getInstance(ALGORITHM);
            cipher.init(Cipher.DECRYPT_MODE, secretKey);
            byte[] decryptedValue = cipher.doFinal(encryptedValue);
            return new String(decryptedValue);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void close() {
        // 清理资源
    }
}

注意事项

兼容性：
- 序列化和反序列化的格式必须严格一致。任何格式的变动都可能导致反序列化失败。例如，如果序列化时动态字段采用JSON格式，反序列化时必须按JSON解析。
- 考虑Kafka版本兼容性。不同版本的Kafka对序列化器和反序列化器的接口可能有细微差异，确保代码在目标Kafka版本上能正常运行。
安全性：
- 加密密钥的管理至关重要。密钥应妥善保存，避免泄露。例如可以使用密钥管理系统（KMS）来存储和管理密钥。
- 选择合适的加密算法和模式。确保加密算法强度足够，能抵御常见的攻击。
性能：
- 序列化和反序列化操作可能会带来性能开销。尽量优化代码，减少不必要的转换和计算。例如，在序列化嵌套对象时，可以采用更高效的数据结构转换方式。
- 避免在序列化和反序列化过程中产生过多的临时对象，以减少垃圾回收的压力。
错误处理：
- 在序列化和反序列化方法中，要妥善处理可能出现的异常。例如反序列化时如果数据格式错误，应抛出合适的异常并记录日志，以便排查问题。
- 考虑数据损坏的情况。如果在传输过程中数据被损坏，反序列化应能检测到并采取相应措施，如丢弃该消息或进行重试。

星途面试题库

面试题：自定义 Kafka 消息序列化与反序列化器实现复杂业务场景需求

知识考点

面试题答案

实现步骤

关键代码逻辑

注意事项