public class Customer {private int customId;private String customerName;public Customer(int customId, String customerName) {this.customId = customId;this.customerName = customerName;}public int getCustomId() {return customId;}public void setCustomId(int customId) {this.customId = customId;}public String getCustomerName() {return customerName;}public void setCustomerName(String customerName) {this.customerName = customerName;}
}

import org.apache.commons.lang.SerializationException;
import org.apache.kafka.common.serialization.Serializer;import java.nio.ByteBuffer;
import java.util.Map;/*** 自定义序列化器* */
public class CustomerSerializer implements Serializer<Customer> {public void configure(Map<String, ?> configs, boolean isKey) {//不做任何配置
    }/*** Customer对象被序列化成:* 1. 表示customerId的4字节整数* 2. 表示customerName长度的4字节整数（如果customerName为空，则长度为0）* 3. 表示customerName的N个字节** */public byte[] serialize(String topic, Customer data) {try{byte[] serializedName;int stringSize;if(data == null)return null;else{if(data.getCustomerName() != null){serializedName = data.getCustomerName().getBytes("UTF-8");stringSize = serializedName.length;} //ifelse{serializedName = new byte[0];stringSize = 0;}ByteBuffer buffer = ByteBuffer.allocate(4 + 4 + stringSize);buffer.putInt(data.getCustomId());buffer.putInt(stringSize);buffer.put(serializedName);return buffer.array();} //else} catch (Exception e){e.printStackTrace();throw new SerializationException("Error when serizlizing custromer to byte[]" +e);}}public void close() {}
}