背景

Flink在消费到Kafka的数据后，我们经常会把数据转成POJO对象，方便后面对数据的处理，但是每个POJO类型是不一样的，每次需要新开发一个反序列化的类，这样就会有大量重复的代码，怎么实现一个通胀的反序列化类解决呢?

另一个场景，当flink同时订阅多个kafka的topic时，我们需要根据不同的topic做不同的逻辑处理，来源topic_A的数据，设置pojo的一个属性为1，来源topic_B的数据，设置POJO的属性为2，这要怎么处理呢？

方法

以上两个问题其实都只要实现KafkaDeserializationSchema这个接口就可以解决问题。

1、通用型反序列化

/** * @desc 自定义实现kafka的消息反序列化 * @throws */public class CustomKafkaDeserializationSchema<T>  implements KafkaDeserializationSchema<T> {    @Override    public boolean isEndOfStream(T nextElement) {        return false;    }    @Override    public T deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {        TypeReference<T> MAP_TYPE_REFERENCE =new TypeReference<T>(){};        T bTrack = JSON.parseObject(record.value(),MAP_TYPE_REFERENCE.getType());        return bTrack;    }    @Override    public TypeInformation<T> getProducedType() {        return TypeInformation.of(new TypeHint<T>() {            @Override            public TypeInformation<T> getTypeInfo() {                return super.getTypeInfo();            }        });    }}

2、自定义序列化

// kafka数据源DataStreamSource<POJO> stream = env.addSource(        new FlinkKafkaConsumer<POJO>(topics, new CustomKafkaDeserializationSchema<POJO>(), PropertyFileUtils.readProFile("kafka.properties"))                .setCommitOffsetsOnCheckpoints(true)                .setStartFromLatest());

3、如果需要获取kafka相关信息，如下图方法就可以读取到