文章目录
- 1. 默认序列化
- 2.自定义序列化
- 3.示例
- 4.自定义解序列化器
1. 默认序列化
在网络中发送数据都是以字节的方式，Kafka也不例外。Apache Kafka支持用户给broker发送各种类型的消息。它可以是一个字符串、一个整数、一个数组或是其他任意的对象类型。序列化器(serializer)负责在producer发送前将消息转换成字节数组;而与之相反，解序列化器(deserializer)则用于将consumer接收到的字节数组转换成相应的对象。
常见的serializer有：
- ByteArraySerializer:本质上什么都不用做，因为已经是字节数组了。
- ByteBufferSerializer:列化ByteBuffer。
- BytesSerializer:序列化Kafka自定义的 Bytes 类。
- DoubleSerializer:列化 Double 类型
- IntegerSerializer:列化Integer 类型
- LongSerializer:序列化Long类型。
- StringSerializer:序列化 String 类型。
producer的序列化机制使用起来非常简单，只需要在构造producer时同时指定参数key.serializer 和 value.serializer的值即可，用户可以为消息的key和value 指定不同类型的 serializer，只要与解序列类型分别保持一致就可以。
2.自定义序列化
Kafka支持用户自定义消息序列化。若要编写一个自定义的serializer,需要完成以下3件事情。
1)定义数据对象格式。
2)创建自定义序列化类，实现 org.apache.kafka.common.serialization.Serializer 接口，在serializer方法中实现序列化逻辑。
3)在用于构造KafkaProducer 的Properties 对象中设置 key.serializer 或 value.serializer取决于是为消息key还是 value 做自定义序列化。
3.示例
下面结合一个实例来说明如何创建自定义的serializer。首先定义待序列化的数据对象。本例中使用一个简单的Java POJO对象，如下面的代码所示:
public class User {private String firstName;private String lastName;private int age;private String address;public User(String firstName, String lastName, int age, String address) {this.firstName = firstName;this.lastName = lastName;this.age = age;this.address = address;}@Overridepublic String toString() {return "User{" +"firstName='" + firstName + '\'' +", lastName='" + lastName + '\'' +", age=" + age +", address='" + address + '\'' +'}';}
}
接下来创建 serializer。本例中使用了jackson-mapper-asl包的 ObjectMapper 帮助我们直接把对象转成字节数组。为了使用该类，你需要在producer工程中增加依赖:
<dependency><groupId>org.codehaus.jackson</groupId><artifactId>jackson-mapper-asl</artifactId><version>1.9.13</version>
</dependency>
UserSerializer代码如下：
import org.apache.kafka.common.serialization.Serializer;
import org.codehaus.jackson.map.ObjectMapper;
import java.io.IOException;
import java.util.Map;
public class UserSerializer implements Serializer<User> {private ObjectMapper objectMapper;@Overridepublic void configure(Map configs, boolean isKey) {objectMapper=new ObjectMapper();}@Overridepublic byte[] serialize(String topic, User data) {byte[] ret =null;try {if (data == null){System.out.println("Null received at serializing");return null;}ret=objectMapper.writeValueAsString(data).getBytes();} catch (IOException e) {e.printStackTrace();}return ret;}@Overridepublic void close() {}
}
指定Serializer，然后构建消息发送：
import com.exm.collectcodenew.kafka.producer.customSerializer.User;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
public class ProducerTest {public static void main(String[] args) throws ExecutionException, InterruptedException {Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");//必须指定props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");//必须指定props.put("value.serializer", "com.exm.collectcodenew.kafka.producer.customSerializer.UserSerializer");//必须指定props.put("acks", "-1");props.put("retries", 3);props.put("batch.size", 323840);props.put("linger.ms", 10);props.put("buffer.memory", 33554432);props.put("max.block.ms", 3000);props.put("partitioner.class","com.exm.collectcodenew.kafka.producer.customPartitioner.AuditPartitioner");Producer<String, String> producer = new KafkaProducer<>(props);//构建User对象User user = new User("Z","tt",18,"Beijing,China");ProducerRecord record = new ProducerRecord("topic-test",user);producer.send(record);producer.close();}
}
4.自定义解序列化器
Kafka支持用户自定义消息的deserializer。成功编写一个自定义的deserializer需要完成以下3件事情。
1）定义或复用 serializer 的数据对象格式,
2) 创建自定义 deserializer 类，令其实现 org.apache.kafka.common.serialization.Deserializer接口。在deserializer方法中实现 deserialize 逻辑。
3）在构造KafkaConsumer的Properties对象中设置key.deserializer和(或)value.deserializer为上一步的实现类。
依然使用序列化中的User 例子来实现自定义的 deserializer。代码如下。
import org.apache.kafka.common.serialization.Deserializer;
import org.codehaus.jackson.map.ObjectMapper;
import java.io.IOException;
import java.util.Map;
public class UserDeserializer implements Deserializer {private ObjectMapper objectMapper;@Overridepublic void configure(Map configs, boolean isKey) {objectMapper = new ObjectMapper();}@Overridepublic Object deserialize(String topic, byte[] data) {User user =null;try {user=objectMapper.readValue(data,User.class);} catch (IOException e) {throw new RuntimeException(e);}finally {return user;}}@Overridepublic void close() {}
}
消费者代码：
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.Properties;
public class ConsumerTest {public static void main(String[] args) {Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");//必须指定props.put("group.id","test-group");//必须指定props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");//必须指定props.put("value.deserializer", "com.exm.collectcodenew.kafka.producer.customSerializer.UserDeserializer");//必须指定props.put("enable.auto.commit","true");props.put("auto.commit.interval.ms","1000");props.put("auto.offset.reset","earliest");//从最早的消息开始读取KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);//创建consumer实例consumer.subscribe(Arrays.asList("topic-test"));while(true){ConsumerRecords<String,String> records=consumer.poll(1000);for (ConsumerRecord<String, String> record: records){System.out.printf("offset=%d,key=%s,value=%s%n",record.offset(),record.key(),record.value());}}}
}
