目录
一、初始化producer对象（序列化消息）
生产者发送消息的三种方式
kafka生产者其它详细知识：
二、初始化consumer对象（反序列化消息）
consumer取消订阅的方式consumer.unsubscribe();
使用自定义的序列化
一、初始化producer对象（序列化消息）
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");
//props.put("value.serializer", "com.java.kafka.PersonUtilSerializer");
//Producer<String, Object> producer = new KafkaProducer<String, Object>(props);
kafkaProducer = new KafkaProducer<>(props);
int i=0;
while(true){
i++;
JsonObject p = new JsonObject();
p .put("id","1")
p .put("name","tester")
System.out.println(p.toString());
// 当指定发送消息的分区时，程序就不会根据key值再判断发往哪个分区了。
// record = new ProducerRecord<>(TOPIC, 0, String.valueOf(i), messageStr);
ProducerRecord<String, Object> record = new ProducerRecord<String,String>(TOPIC, p.toString());
producer.send(record);
}
producer.close();
}
ProducerRecord 构造方法，该方法的参数 topic 和 value 属性是必填项，其余属性（比如：分区号、时间戳、key、headers）是选填项。对应的 ProducerRecord 的构造方法也有多种：使用者可根据场景来选择合适的 ProducerRecord 。
属性key.serializer和value.serializer就是key和value指定的序列化方式。无论是key还是value序列化和反序列化实现都是一样的。
StringSerializer是内置的字符串序列化方式
生产者发送消息的三种方式
Kafka 生产者发送消息有三种方式，分别为：普通发送（发后即忘）、同步发送、异步发送。
kafka生产者其它详细知识：
Kafka基础（二）：生产者相关知识汇总
https://blog.csdn.net/CREATE_17/article/details/93396981
二、初始化consumer对象（反序列化消息）
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");
props.put("group.id", "group1");
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
// 订阅test1 topic 使用subscribe()方法订阅主题 （一般推荐这种方式）
consumer.subscribe(Collections.singletonList("test1"));
// 另一种订阅的方式 使用assign()方法订阅确定主题和分区
//consumer.assign(Collections.singletonList(new TopicPartition(partitionInfo.topic(), partitionInfo.partition())));
consumer.seekToBeginning(Arrays.asList(TOPIC_PARTITION));
while(true){
ConsumerRecords<String, String> records = consumer.poll(100);
System.err.println("print the size of records ,size="+records.count());
for(ConsumerRecord<String, String> record:records){
System.out.println(record);
}
}
属性key.deserializer和value.deserializer就是key和value指定的反序列化方式。
StringDeserializer是内置的字符串反序列化方式
consumer取消订阅的方式consumer.unsubscribe();
可以使用 unsubscribe() 方法来取消主题的订阅
使用方式：consumer.unsubscribe();
unsubscribe()方法即可以取消通过subscribe()方式实现的订阅，还可以取消通过assign()方式实现的订阅
consumer的两种不同订阅方式：subscribe() 和 assign()
使用自定义的序列化
写一个自定义的类，比如说我们要传递一个Person对象，那么我们就定义个Person对象的序列化和反序列化的类，并且实现Serializer接口，下面继续看，首先定义个Person类
public class Person implements Serializable{
private String id ;
private String name ;
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
@Override
public String toString() {
return "Person [id=" + id + ", name=" + name + "]";
}
}
接下来，我们自定义一个序列化的类： PersonUtilSerializer
public class PersonUtilSerializer implements Serializer<Person>{
@Override
public void configure(Map<String, ?> configs, boolean isKey) {
}
@Override
public byte[] serialize(String topic, Person data) {
return JSON.toJSONBytes(data);
}
@Override
public void close() {
// TODO Auto-generated method stub
}
}
初始化producer实例对象
props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "com.java.kafka.PersonUtilSerializer");
Producer<String, Object> producer = new KafkaProducer<String, Object>(props);
int i=0;
while(true){
i++;
Person p = new Person();
p.setId(i+"");
p.setName("zhangsan-"+i);
System.out.println(p.toString());
ProducerRecord<String, Object> record = new ProducerRecord<String,Object>(TOPIC, p);
producer.send(record);
}
producer.close();
}
自定义Serializer和Deserializer非常痛苦，而且上面还有很多异常情况没有处理，还有很多类型不支持，非常脆弱。复杂类型的支持更是一件痛苦的事情，不同版本之间的兼容性问题更是一个极大的挑战。由于Serializer和Deserializer影响到上下游系统，导致牵一发而动全身。自定义序列化&反序列化实现不是能力的体现，而是逗比的体现。所以强烈不建议自定义实现序列化&反序列化，推荐直接使用StringSerializer和StringDeserializer，然后使用json作为标准的数据传输格式。站在巨人的肩膀上，事半功倍。
Kafka中位移提交那些事儿
https://blog.csdn.net/weixin_45039616/article/details/107081148
