自定义kafka序列化和反序列化器

### 自定义 Kafka 序列化器和反序列化器 为了实现自定义的 Kafka 序列化器和反序列化器，可以遵循以下方式： #### 实现自定义序列化器 Kafka 的 `Serializer` 接口提供了两个核心方法：`configure()` 和 `serialize()`。以下是基于 Java 的一个简单示例代码，展示如何创建一个用于对象 `Company` 的自定义序列化器。 ```java import org.apache.kafka.common.serialization.Serializer; import java.nio.ByteBuffer; import java.util.Map; public class CustomSerializer implements Serializer<Company> { @Override public void configure(Map<String, ?> configs, boolean isKey) { // 配置逻辑可在此处编写 } @Override public byte[] serialize(String topic, Company data) { if (data == null) { return null; } try { ByteBuffer buffer = ByteBuffer.allocate(1024); buffer.put(data.getName().getBytes()); buffer.put(data.getAddress().getBytes()); buffer.flip(); byte[] result = new byte[buffer.limit()]; buffer.get(result); return result; } catch (Exception e) { throw new RuntimeException("Error serializing object", e); } } @Override public void close() { // 清理资源（如果有） } } ``` 此代码实现了将 `Company` 对象转换为字节数组的功能[^2]。 --- #### 实现自定义反序列化器 同样地，可以通过实现 `Deserializer` 接口来完成反序列化的操作。下面是对应的反序列化器代码示例： ```java import org.apache.kafka.common.serialization.Deserializer; import java.nio.ByteBuffer; import java.util.Map; public class CustomDeserializer implements Deserializer<Company> { @Override public void configure(Map<String, ?> configs, boolean isKey) { // 可配置初始化参数 } @Override public Company deserialize(String topic, byte[] data) { if (data == null || data.length == 0) { return null; } try { ByteBuffer buffer = ByteBuffer.wrap(data); int nameLength = buffer.getInt(); // 假设名称长度存储在前四个字节中 byte[] nameBytes = new byte[nameLength]; buffer.get(nameBytes); String name = new String(nameBytes); int addressLength = buffer.getInt(); // 地址长度假设也存于固定位置 byte[] addressBytes = new byte[addressLength]; buffer.get(addressBytes); String address = new String(addressBytes); return new Company(name, address); } catch (Exception e) { throw new RuntimeException("Error deserializing object", e); } } @Override public void close() { // 关闭清理工作 } } ``` 这段代码展示了如何从字节数组还原成原始的对象实例[^4]。 --- #### 使用自定义序列化器和反序列化器 当生产者发送消息时，需指定使用的序列化器；消费者接收消息时则应设置相应的反序列化器。例如，在 Kafka 生产者的配置文件中添加如下内容即可启用自定义序列化器： ```properties key.serializer=org.apache.kafka.common.serialization.StringSerializer value.serializer=com.example.CustomSerializer ``` 对于消费者的配置，则需要指明所用的反序列化器类名： ```properties key.deserializer=org.apache.kafka.common.serialization.StringDeserializer value.deserializer=com.example.CustomDeserializer ``` 以上步骤确保了数据能够在生产和消费过程中被正确处理[^3]。 ---