Kafka Stream简单示例（四）---定义更通用的Serde

Dear 丶 2022-03-10 07:50 245阅读 0赞

本篇是在[《Kafka Stream简单示例（一）》][Kafka Stream] 和[《Kafka Stream简单示例（二）—聚合 Aggregation–统计总和》][Kafka Stream_ Aggregation] 以及[《 Kafka Stream简单示例（三）—自定义Serde》][Kafka Stream_Serde]基础上成文的，建议先阅读前三篇，以便清楚上下文关系需求背景。

第三篇 《Kafka Stream简单示例（三）—自定义Serde》中，我们自定义了Statistic类的Serializer和Deserializer。现实中我们可能需要多个类都支持序列化和反序列，能否有泛型的Serializer和Deserializer，直接放入自己的类就可以完成工作？答案是，如果你的类，是POJO类型的，使用泛型JsonPOJOSerializer和JsonPOJODeserializer就可以。

**注意**：示例中的代码只是展示流程，非生产代码，仅供参考，由此导致的问题本人概不负责。

官方文档在这里，我用是kafka 1.0. 所以连接也是1.0版本的文档。 [http://kafka.apache.org/10/documentation/streams/developer-guide/datatypes.html][http_kafka.apache.org_10_documentation_streams_developer-guide_datatypes.html]

# 项目需求 #

统计一分钟内（固定时间窗口Tumbling Window）内温度的总和与平均值。类似的还有，最大值，最小值。

# 主要流程和代码 #

完整的代码在[这里][Link 1]，欢迎加星和fork。 谢谢！

一个结果中必须同时含有总和与平均值，于是我们设计一个简单数据结构

@Data
    @AllArgsConstructor
    @NoArgsConstructor
    public class Statistics {
        private Long avg;
        private Long sum;
        private Long count;
    }

细心的人会发现本篇中的Statistic比[《 Kafka Stream简单示例（三）—自定义Serde》][Kafka Stream_Serde]中的Statistic多一个@NoArgsConstructor注解，这是因为我们后面使用反序列化是需要生成Statistics对象（使用默认的无参构造函数生成）。 因此需要添加@NoArgsConstructor注解。

根据Serdes的要求，我们必须提供对应的Serializer和Deserializer。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1c3NsZQ_size_16_color_FFFFFF_t_70]

我们需要实现JsonPOJOSerializer和JsonPOJODeserializer。仍然才考LongSerializer和LongDeserializer的实现， 我们实现了StatisticsSerializer和StatisticsDeserializer。  
**首先是序列化实现JsonPOJOSerializer**

package com.yq.generic;
    
    import com.fasterxml.jackson.databind.ObjectMapper;
    import org.apache.kafka.common.serialization.Serializer;
    import org.apache.kafka.common.errors.SerializationException;
    
    import java.util.Map;
    
    /** * 这个是官方例子的copy， 版权归官方。copy到本地是为了让我的例子也运行起来 * https://github.com/apache/kafka/blob/1.0/streams/examples/src/main/java/org/apache/kafka/streams/examples/pageview/JsonPOJOSerializer.java * className: JsonPOJOSerializer * */
    
    
    public class JsonPOJOSerializer<T> implements Serializer<T> { 
        private final ObjectMapper objectMapper = new ObjectMapper();
    
        /** * Default constructor needed by Kafka */
        public JsonPOJOSerializer() { 
        }
    
        @Override
        public void configure(Map<String, ?> props, boolean isKey) { 
        }
    
        @Override
        public byte[] serialize(String topic, T data) { 
            if (data == null)
                return null;
    
            try { 
                return objectMapper.writeValueAsBytes(data);
            } catch (Exception e) { 
                throw new SerializationException("Error serializing JSON message", e);
            }
        }
    
        @Override
        public void close() { 
        }
    
    }

**其次是反序列化实现JsonPOJODeserializer**

package com.yq.generic;
    
    import com.fasterxml.jackson.databind.ObjectMapper;
    import org.apache.kafka.common.serialization.Deserializer;
    import org.apache.kafka.common.errors.SerializationException;
    
    import java.util.Map;
    
    /** * 这个是官方例子的copy， 版权归官方。copy到本地是为了让我的例子也运行起来 * https://github.com/apache/kafka/blob/1.0/streams/examples/src/main/java/org/apache/kafka/streams/examples/pageview/JsonPOJODeserializer.java * className: JsonPOJOSerializer * */
    public class JsonPOJODeserializer<T> implements Deserializer<T> { 
        private ObjectMapper objectMapper = new ObjectMapper();
    
        private Class<T> tClass;
    
        /** * Default constructor needed by Kafka */
        public JsonPOJODeserializer() { 
        }
    
        @SuppressWarnings("unchecked")
        @Override
        public void configure(Map<String, ?> props, boolean isKey) { 
            tClass = (Class<T>) props.get("JsonPOJOClass");
        }
    
        @Override
        public T deserialize(String topic, byte[] bytes) { 
            if (bytes == null)
                return null;
    
            T data;
            try { 
                data = objectMapper.readValue(bytes, tClass);
            } catch (Exception e) { 
                throw new SerializationException(e);
            }
    
            return data;
        }
    
        @Override
        public void close() { 
    
        }
    }

**最后是我们的主流程。**  
第一步，我们需要先定义

final Serializer<Statistics> statisticsSerializer = new JsonPOJOSerializer<>();
            serdeProps.put("JsonPOJOClass", Statistics.class);
            statisticsSerializer.configure(serdeProps, false);
    
            final Deserializer<Statistics> statisticsDeserializer = new JsonPOJODeserializer<>();
            serdeProps.put("JsonPOJOClass", Statistics.class);
            statisticsDeserializer.configure(serdeProps, false);
    
            final Serde<Statistics> statisticsSerde = Serdes.serdeFrom(statisticsSerializer, statisticsDeserializer);

第二步。 就像SerDe内置的Serdes.Long()或者 Serdes.String()， 可以直接使用statisticsSerde。

KTable的格式是 KTable<Windowed, Statistics>。 aggregate函数的初始值和返回都是Statistics类型， 结果存储的格式Materialized.<String, Statistics, WindowStore<Bytes, byte\[\]>>as(“time-windowed-aggregated-temp-stream-store”)  
.withValueSerde(statisticsSerde) ， 也是Statistics类型。

package com.yq.generic;
    
    
    import com.alibaba.fastjson.JSON;
    import com.alibaba.fastjson.JSONObject;
    import com.yq.customized.Statistics;
    import org.apache.kafka.clients.consumer.ConsumerConfig;
    import org.apache.kafka.common.serialization.Deserializer;
    import org.apache.kafka.common.serialization.Serde;
    import org.apache.kafka.common.serialization.Serdes;
    import org.apache.kafka.common.serialization.Serializer;
    import org.apache.kafka.common.utils.Bytes;
    import org.apache.kafka.streams.KafkaStreams;
    import org.apache.kafka.streams.StreamsBuilder;
    import org.apache.kafka.streams.StreamsConfig;
    import org.apache.kafka.streams.kstream.Aggregator;
    import org.apache.kafka.streams.kstream.Initializer;
    import org.apache.kafka.streams.kstream.KStream;
    import org.apache.kafka.streams.kstream.KTable;
    import org.apache.kafka.streams.kstream.KeyValueMapper;
    import org.apache.kafka.streams.kstream.Materialized;
    import org.apache.kafka.streams.kstream.Produced;
    import org.apache.kafka.streams.kstream.TimeWindows;
    import org.apache.kafka.streams.kstream.Windowed;
    import org.apache.kafka.streams.kstream.internals.WindowedDeserializer;
    import org.apache.kafka.streams.kstream.internals.WindowedSerializer;
    import org.apache.kafka.streams.state.WindowStore;
    
    import java.util.HashMap;
    import java.util.Map;
    import java.util.Properties;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.TimeUnit;
    
    /** * http://kafka.apache.org/10/documentation/streams/developer-guide/datatypes.html * 统计60秒内，温度值的最大值 topic中的消息格式为数字，30， 21或者{"temp":19, "humidity": 25} */
    public class TemperatureAvgGenericSerDeDemo { 
        private static final int TEMPERATURE_WINDOW_SIZE = 60;
    
        public static void main(String[] args) throws Exception { 
    
            Properties props = new Properties();
            props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-temp-avg");
            props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
            props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
            props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
    
            StreamsBuilder builder = new StreamsBuilder();
    
            KStream<String, String> source = builder.stream("iot-temp");
    
            Map<String, Object> serdeProps = new HashMap<>();
    
            final Serializer<Statistics> statisticsSerializer = new JsonPOJOSerializer<>();
            serdeProps.put("JsonPOJOClass", Statistics.class);
            statisticsSerializer.configure(serdeProps, false);
    
            final Deserializer<Statistics> statisticsDeserializer = new JsonPOJODeserializer<>();
            serdeProps.put("JsonPOJOClass", Statistics.class);
            statisticsDeserializer.configure(serdeProps, false);
    
            final Serde<Statistics> statisticsSerde = Serdes.serdeFrom(statisticsSerializer, statisticsDeserializer);
    
            KTable<Windowed<String>, Statistics> max = source
                    .selectKey(new KeyValueMapper<String, String, String>() { 
                        @Override
                        public String apply(String key, String value) { 
                            return "stat";
                        }
                    })
                    .groupByKey()
                    .windowedBy(TimeWindows.of(TimeUnit.SECONDS.toMillis(TEMPERATURE_WINDOW_SIZE)))
                    .aggregate(
                            new Initializer<Statistics>() { 
                                @Override
                                public Statistics apply() { 
                                    Statistics avgAndSum = new Statistics(0L,0L,0L);
                                    return avgAndSum;
                                }
                            },
                            new Aggregator<String, String, Statistics>() { 
                                @Override
                                public Statistics apply(String aggKey, String newValue, Statistics aggValue) { 
                                    //topic中的消息格式为{"temp":19, "humidity": 25}
                                    System.out.println("aggKey:" + aggKey + ", newValue:" + newValue + ", aggKey:" + aggValue);
                                    Long newValueLong = null;
                                    try { 
                                        JSONObject json = JSON.parseObject(newValue);
                                        newValueLong = json.getLong("temp");
                                    }
                                    catch (ClassCastException ex) { 
                                        try { 
                                            newValueLong = Long.valueOf(newValue);
                                        }
                                         catch (NumberFormatException e) { 
                                             System.out.println("Exception:" + e.getMessage());
                                             //异常返回原值
                                             return aggValue;
                                        }
                                    }
                                    catch (Exception e) { 
                                        System.out.println("Exception:" + e.getMessage());
                                        //异常返回原值
                                        return aggValue;
                                    }
    
    
                                    aggValue.setCount(aggValue.getCount() + 1);
                                    aggValue.setSum(aggValue.getSum() + newValueLong);
                                    aggValue.setAvg(aggValue.getSum() / aggValue.getCount());
    
                                    return aggValue;
                                }
                            },
                            Materialized.<String, Statistics, WindowStore<Bytes, byte[]>>as("time-windowed-aggregated-temp-stream-store")
                                    .withValueSerde(statisticsSerde)
                    );
    
            WindowedSerializer<String> windowedSerializer = new WindowedSerializer<>(Serdes.String().serializer());
            WindowedDeserializer<String> windowedDeserializer = new WindowedDeserializer<>(Serdes.String().deserializer(), TEMPERATURE_WINDOW_SIZE);
            Serde<Windowed<String>> windowedSerde = Serdes.serdeFrom(windowedSerializer, windowedDeserializer);
    
            max.toStream().to("iot-temp-stat", Produced.with(windowedSerde, statisticsSerde));
    
            final KafkaStreams streams = new KafkaStreams(builder.build(), props);
            final CountDownLatch latch = new CountDownLatch(1);
    
    
            Runtime.getRuntime().addShutdownHook(new Thread("streams-temperature-shutdown-hook") { 
                @Override
                public void run() { 
                    streams.close();
                    latch.countDown();
                }
            });
    
            try { 
                streams.start();
                latch.await();
            } catch (Throwable e) { 
                System.exit(1);
            }
            System.exit(0);
        }
    }

# 效果截图 #

图中已经有文字说明，结合代码能更清楚了解Kafka Stream。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1c3NsZQ_size_16_color_FFFFFF_t_70 1]

[Kafka Stream]: https://blog.csdn.net/russle/article/details/88091448
[Kafka Stream_ Aggregation]: https://blog.csdn.net/russle/article/details/88360685
[Kafka Stream_Serde]: https://blog.csdn.net/russle/article/details/88366093
[http_kafka.apache.org_10_documentation_streams_developer-guide_datatypes.html]: http://kafka.apache.org/10/documentation/streams/developer-guide/datatypes.html
[Link 1]: https://github.com/yqbjtu/KafkaTutorials/tree/master/KafkaStreamDemo
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1c3NsZQ_size_16_color_FFFFFF_t_70]: /images/20220310/7210711d6b8c45a9a233e07d93bec19b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1c3NsZQ_size_16_color_FFFFFF_t_70 1]: /images/20220310/85546ab2f593448ebba055cd7479bff7.png