序列化
kafka序列化消息是在生产端，序列化后，消息才能网络传输。而构造KafkaProducer代码如下：
Properties props = new Properties();
props.put("bootstrap.servers", "10.0.55.229:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
kafkaProducer = new KafkaProducer<>(props);
属性key.serializer和value.serializer就是key和value指定的序列化方式。无论是key还是value序列化和反序列化实现都是一样的，所以接下来都只以value的序列化和反序列为例。
StringSerializer
StringSerializer是内置的字符串序列化方式，核心源码如下：
/**
* String encoding defaults to UTF8 and can be customized by setting the property key.serializer.encoding,
* value.serializer.encoding or serializer.encoding. The first two take precedence over the last.
*/
public class StringSerializer implements Serializer<String> {
private String encoding = "UTF8";
... ...
@Override
public byte[] serialize(String topic, String data) {
try {
// 如果数据为空，那么直接返回null即可
if (data == null)
return null;
else
// 否则将String序列化，即转为byte[]即可
return data.getBytes(encoding);
} catch (UnsupportedEncodingException e) {
throw new SerializationException("Error when serializing string to byte[] due to unsupported encoding " + encoding);
}
}
@Override
public void close() {
// nothing to do
}
}
自定义序列化
和内置的StringSerializer字符串序列化一样，如果要自定义序列化方式，需要实现接口Serializer。假设每个字段按照下图所示的方式自定义序列化：
image.png
下面是一个简单的自定义实现（假设自定义Order类型的Serializer，Order类中有一些String，Integer，Long，Date类型的属性--其他类型暂不支持，读者可以自行扩展）：
/**
* @author wangzhenfei9
* @version 1.0.0
* @since 2018年06月22日
*/
public class OrderSerializer implements Serializer<Order> {
@Override
public void configure(Map<String, ?> configs, boolean isKey) {
// nothing to do
}
@Override
public byte[] serialize(String topic, Order data){
if (data == null) {
return null;
}
Class<? extends Order> clazz = data.getClass();
Field[] fields = clazz.getDeclaredFields();
// 计算保存Order所示属性值总计需要多少字节
int total = 0;
// 遍历所有属性字段, 目前只支持Integer, Long, String, Date类型
for (Field field:fields){
String fieldName = field.getName();
int fieldLen = fieldName.getBytes().length;
Object value = getPropertyValue(data, fieldName);
String type = field.getType().getSimpleName();
System.out.println("propertyName: "+fieldName +", value: " + value
+", len: " + fieldLen+", type: " + type);
// 每个属性序列化方式: 属性长度(都是4个字节)+属性名称(长度需要计算)+值长度+值
switch (type){
case "Long":
// 第一个4表示属性名长度需要的空间, 即int的长度；（int类型反序列化后需要4个字节长度，即32位）
// 第二个fieldLen表示属性名需要的空间
// 第三个4表示属性值长度需要的空间
// 第四个8表示属性值需要的空间（Long类型反序列化后需要8个字节长度，即64位）
total+=(4+fieldLen+4+8);
break;
case "Date":
// +8+8
// 如果是日期类型先转成Long类型的timestamp
total+=(4+fieldLen+4+8);
break;
case "Integer":
total+=(4+fieldLen+4+4);
break;
case "String":
try {
total+=(4+fieldLen + 4 + value.toString().getBytes("utf-8").length);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
break;
// 本次自定义的Serializer只支持Integer, Long, Date, String类型属性, 如果想要支持其他类型属性, 请自行实现
default:
throw new IllegalArgumentException("Unsupported argument type: "+type);
}
}
// 算出需要个byte[]数组长度后, 分配同样大的byte[]中
ByteBuffer result = ByteBuffer.allocate(total);
for (Field field:fields){
String fieldName = field.getName();
byte[] fieldByte = fieldName.getBytes();
int fieldLen = fieldByte.length;
Object value = getPropertyValue(data, fieldName);
String type = field.getType().getSimpleName();
// 无论什么类型都要先put属性长度和属性名称
result.putInt(fieldLen);
result.put(fieldByte);
switch (type){
case "Long":
byte[] longByte = SerializeUtil.longSerialize((Long) value);
result.putInt(longByte.length);
result.put(longByte);
break;
case "Date":
// 如果是日期类型先转成Long类型的timestamp
byte[] dateByte = SerializeUtil.longSerialize(((Date) value).getTime());
result.putInt(dateByte.length);
result.put(dateByte);
break;
case "Integer":
byte[] integerByte = SerializeUtil.integerSerialize((Integer) value);
result.putInt(integerByte.length);
result.put(integerByte);
break;
case "String":
byte[] stringByte = ((String)value).getBytes();
result.putInt(stringByte.length);
result.put(stringByte);
break;
default:
throw new IllegalArgumentException("Unsupported argument type: "+type);
}
}
// 返回序列化后的结果
return result.array();
}
private Object getPropertyValue(Order order, String propertyName) {
Method[] methods = order.getClass().getMethods();
for (Method method : methods) {
// 这里方法匹配还不够严谨
if (method.getName().equalsIgnoreCase("get" + propertyName)
|| method.getName().equalsIgnoreCase("is" + propertyName)) {
try {
return method.invoke(order);
} catch (Exception e) {
e.printStackTrace();
}
}
}
return null;
}
@Override
public void close() {
// nothing to do
}
}
自定义Serializer后，修改属性value.serializer的值为com.afei.kafka.serialization.OrderSerializer，且ProducerRecord申明为ProducerRecord<String, Order>。
反序列化
kafka反序列化消息是在消费端。由于网络传输过来的是byte[]，只有反序列化后才能得到生产者发送的真实的消息内容。而构造KafkaConsumer代码如下：
Properties props = new Properties();
props.put("bootstrap.servers", "10.0.55.229:9092");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
属性key.deserializer和value.deserializer就是key和value指定的反序列化方式。
StringDeserializer
StringDeserializer是内置的字符串反序列化方式，核心源码如下：
/**
* String encoding defaults to UTF8 and can be customized by setting the property key.deserializer.encoding,
* value.deserializer.encoding or deserializer.encoding. The first two take precedence over the last.
*/
public class StringDeserializer implements Deserializer<String> {
private String encoding = "UTF8";
... ...
@Override
public String deserialize(String topic, byte[] data) {
try {
// 如果数据为空，那么直接返回null即可
if (data == null)
return null;
else
// 否则将byte[]反序列化，即转为String即可
return new String(data, encoding);
} catch (UnsupportedEncodingException e) {
throw new SerializationException("Error when deserializing byte[] to string due to unsupported encoding " + encoding);
}
}
... ...
}
自定义反序列化
和内置的StringDeserializer字符串反序列化一样，如果要自定义反序列化方式，需要实现接口Deserializer。下面是一个简单的自定义实现（假设自定义Order类型的Deserializer，Order类中有一些String，Integer，Long，Date类型的属性--其他类型暂不支持），反序列化就是根据序列化的方式得到序列化前的内容：
/**
* @author wangzhenfei9
* @version 1.0.0
* @since 2018年06月22日
*/
public class OrderDeserializer implements Deserializer<Order> {
@Override
public void configure(Map<String, ?> configs, boolean isKey) {
// nothing to do
}
@Override
public Order deserialize(String topic, byte[] bytes) {
ByteBuffer data = ByteBuffer.wrap(bytes);
Field[] declaredFields = new Field[0];
try {
declaredFields = Class.forName(Order.class.getCanonicalName()).getDeclaredFields();
} catch (ClassNotFoundException e) {
e.printStackTrace();
}
int propertyCount = declaredFields.length;
Order order = new Order();
while ((propertyCount--)>0){
int propertyLen = data.getInt();
byte[] nameByte = new byte[propertyLen];
data.get(nameByte);
String propertyName = new String(nameByte);
//根据属性名称得到属性的类型
String propertyType = getPropertyType(Order.class, propertyName);
int valueLen = data.getInt();
byte[] valueByte = new byte[valueLen];
data.get(valueByte);
Object value;
switch (propertyType ){
case "Long":
value = DeserializeUtil.longDeserialize(valueByte);
break;
case "Date":
// 如果是日期类型先反序列化成Long类型, 然后得到Date类型的值
// result.putLong(((Date) value).getTime());
value = new Date(DeserializeUtil.longDeserialize(valueByte));
break;
case "Integer":
value = DeserializeUtil.integerDeserialize(valueByte);
break;
case "String":
value = new String(valueByte);
break;
default:
throw new IllegalArgumentException("Unsupported argument type: "+propertyType);
}
setPropertyValue(order, propertyName, value);
System.out.println("property = "+propertyLen+", name = "+propertyName
+", value = "+valueLen+", name = "+value);
}
return order;
}
private String getPropertyType(Class<? extends Order> clazz, String propertyName) {
/*
* 得到类中的方法
*/
try {
Field field = clazz.getDeclaredField(propertyName);
return field.getType().getSimpleName();
} catch (NoSuchFieldException e) {
}
return null;
}
/**
* 调用Order中属性${propertyName}的setter方法并赋值${propertyValue}
*/
private void setPropertyValue(Order order, String propertyName, Object propertyValue) {
/*
* 得到类中的方法
*/
Method[] methods = order.getClass().getMethods();
for (Method method : methods) {
// 这里方法匹配还不够严谨
if (method.getName().equalsIgnoreCase("set" + propertyName)) {
try {
method.invoke(order, propertyValue);
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
@Override
public void close() {
// nothing to do
}
}
自定义Serializer后，修改属性value.serializer的值为com.afei.kafka.serialization.OrderDeserializer，且KafkaConsumer申明为KafkaConsumer<String, Order>。
总结
可以看到，自定义Serializer和Deserializer非常痛苦，而且上面还有很多异常情况没有处理，还有很多类型不支持，非常脆弱。复杂类型的支持更是一件痛苦的事情，不同版本之间的兼容性问题更是一个极大的挑战。由于Serializer和Deserializer影响到上下游系统，导致牵一发而动全身。自定义序列化&反序列化实现不是能力的体现，而是逗比的体现。所以强烈不建议自定义实现序列化&反序列化，推荐直接使用StringSerializer和StringDeserializer，然后使用json作为标准的数据传输格式。站在巨人的肩膀上，事半功倍。
