kafka-examples 集成Avro与Schema RegistryKafkaAvroSerializer序列化全流程实战指南【免费下载链接】kafka-examplesSnippets and small examples demonstrating kafka features and configs项目地址: https://gitcode.com/gh_mirrors/kaf/kafka-exampleskafka-examples 是一个演示 Kafka 特性与配置的实战代码仓库其中 Avro 示例模块完整展示了如何集成 Avro 序列化与 Schema Registry使用 KafkaAvroSerializer 将强类型 Avro 事件写入 Kafka再用 KafkaAvroDeserializer 在消费端自动还原。本文面向新手带你走通定义 Schema → 生产 → 消费的完整链路。 为什么 Kafka 需要 Avro 序列化直接把 JSON 字符串丢进 Kafka 虽然简单但在生产环境中会有三大痛点体积大JSON 每条消息都要重复携带字段名带宽和存储成本更高无契约字段名写错、类型不匹配只有消费时才爆炸演进困难生产者加了新字段消费者老代码直接报错Avro 用二进制格式 中心化 Schema 管理解决了这些问题Schema 统一注册到 Schema Registry消息体里只存一个 5 字节的 Schema ID消费者凭 ID 取回 Schema 自动反序列化。kafka-examples仓库用一对点击流生成器 会话化消费者把这套机制讲得明明白白。 环境准备一键搭建运行环境运行 Avro 示例需要三个组件全部使用默认配置启动即可ZookeeperKafka BrokerSchema RegistryConfluent 提供默认监听http://localhost:8081克隆仓库并构建生产者git clone https://gitcode.com/gh_mirrors/kaf/kafka-examples cd kafka-examples/AvroProducerExample mvn clean package然后创建主题clicksbin/kafka-topics.sh --create --zookeeper localhost:2181 \ --replication-factor 1 --partitions 1 --topic clicks 依赖版本参考 AvroProducerExample/pom.xmlKafka 2.4.0、Confluent 5.4.0、Avro 1.7.7其中kafka-avro-serializer是核心依赖。 定义 Avro Schema一切从 LogLine.avsc 开始项目的数据模型是一个网页点击日志LogLineSchema 定义在AvroProducerExample/src/main/resources/avro/LogLine.avsc{ namespace: JavaSessionize.avro, type: record, name: LogLine, fields: [ {name: ip, type: string}, {name: timestamp, type: long}, {name: url, type: string}, {name: referrer, type: string}, {name: useragent, type: string}, {name: sessionid, type: [null,int], default: null} ] }注意namespacename组合成了 Schema Registry 中的全限定主题名JavaSessionize.avro.LogLine这是 Schema 注册与查找的关键标识。构建时avro-maven-plugin会在generate-sources阶段基于该 Schema自动生成 Java 类Specific 类代码里直接new LogLine()填充字段即可无需手写序列化逻辑。 第一步生产者——用 KafkaAvroSerializer 发送 Avro 事件核心代码在AvroProducerExample/src/main/java/com/shapira/examples/producer/avroclicks/AvroClicksProducer.java配置 Producer 只需关注 4 个参数props.put(key.serializer, io.confluent.kafka.serializers.KafkaAvroSerializer); props.put(value.serializer, io.confluent.kafka.serializers.KafkaAvroSerializer); props.put(schema.registry.url, schemaUrl); props.put(acks, all);EventGenerator同目录EventGenerator.java负责生成模拟点击事件生产者的发送逻辑非常直观LogLine event EventGenerator.getNext(); ProducerRecordString, LogLine record new ProducerRecord(clicks, event.getIp().toString(), event); producer.send(record).get();幕后发生了什么KafkaAvroSerializer首次序列化LogLine时向 Schema Registry 注册其 Schema 并拿到 ID消息体被编码为1 字节魔数 4 字节 Schema ID Avro 二进制数据同一 IP 的事件因为 Key 相同会被路由到同一分区——天然保证了同一用户的事件有序运行生产者写入 100 条点击java -cp target/uber-ClickstreamGenerator-1.0-SNAPSHOT.jar \ com.shapira.examples.producer.avroclicks.AvroClicksProducer 100 http://localhost:8081 第二步消费者——KafkaAvroDeserializer 自动反序列化消费端示例在AvroConsumerExample/src/main/java/com/shapira/examples/consumer/avroclicks/AvroClicksSessionizer.java配置与生产端对称props.put(key.deserializer, org.apache.kafka.common.serialization.StringDeserializer); props.put(value.deserializer, io.confluent.kafka.serializers.KafkaAvroDeserializer); props.put(schema.registry.url, url); props.put(specific.avro.reader, true);关键配置速查表配置项作用示例值key.serializer/value.serializer生产端序列化器KafkaAvroSerializerkey.deserializer/value.deserializer消费端反序列化器KafkaAvroDeserializerschema.registry.urlSchema Registry 地址http://localhost:8081specific.avro.reader用生成的 Specific 类读取trueauto.offset.reset无消费位点时的起点earliestspecific.avro.reader true表示消费时直接还原为编译期生成的LogLine类而不是通用的GenericRecord——类型安全且无需手动取字段。该消费者读取clicks主题后做了一件典型的事会话化。它用内存表记录每个 IP 的最后活跃时间间隔超过 30 分钟就递增sessionid状态管理见同目录SessionState.java再把带会话 ID 的事件写入sessionized_clicks主题——正是消费 → 加工 → 再生产的经典管道模式。运行前先创建输出主题bin/kafka-topics.sh --create --zookeeper localhost:2181 \ --replication-factor 1 --partitions 1 --topic sessions⚠️ 新手避坑清单Schema Registry 必须先启动生产者启动时会连接它注册 Schema连不上会直接抛异常namespace别乱改它决定 Schema 在 Registry 中的全名生产与消费两端必须一致Key 的序列化器是独立的本例 Key 是 IP 字符串用StringSerializer即可无需套用 Avro手动提交位点消费端关闭了自动提交auto.commit.enablefalse处理完一批后commitSync()保证不丢消息验证结果用 Confluent 自带的 Avro 控制台消费者查看落库数据bin/kafka-avro-console-consumer --zookeeper localhost:2181 --topic sessionized_clicks --from-beginning 总结通过 kafka-examples 的 Avro 生产/消费示例你掌握了完整的 Avro 序列化链路用LogLine.avsc定义 SchemaMaven 插件自动生成 Java 类生产端配置KafkaAvroSerializerschema.registry.url完成注册与编码消费端配置KafkaAvroDeserializerspecific.avro.reader自动还原强类型对象借助 Schema Registry 实现 Schema 的集中管理与版本演进这套生产 消费 Schema 治理的组合拳正是企业级 Kafka 数据管道的标准姿势。建议继续阅读仓库中的 AvroProducerExample/README.md 和 AvroConsumerExample/README.md动手跑通后再挑战 Kafka Streams 相关示例进阶流式计算 【免费下载链接】kafka-examplesSnippets and small examples demonstrating kafka features and configs项目地址: https://gitcode.com/gh_mirrors/kaf/kafka-examples创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
kafka-examples 集成Avro与Schema RegistryKafkaAvroSerializer序列化全流程实战指南【免费下载链接】kafka-examplesSnippets and small examples demonstrating kafka features and configs项目地址: https://gitcode.com/gh_mirrors/kaf/kafka-exampleskafka-examples 是一个演示 Kafka 特性与配置的实战代码仓库其中 Avro 示例模块完整展示了如何集成 Avro 序列化与 Schema Registry使用 KafkaAvroSerializer 将强类型 Avro 事件写入 Kafka再用 KafkaAvroDeserializer 在消费端自动还原。本文面向新手带你走通定义 Schema → 生产 → 消费的完整链路。 为什么 Kafka 需要 Avro 序列化直接把 JSON 字符串丢进 Kafka 虽然简单但在生产环境中会有三大痛点体积大JSON 每条消息都要重复携带字段名带宽和存储成本更高无契约字段名写错、类型不匹配只有消费时才爆炸演进困难生产者加了新字段消费者老代码直接报错Avro 用二进制格式 中心化 Schema 管理解决了这些问题Schema 统一注册到 Schema Registry消息体里只存一个 5 字节的 Schema ID消费者凭 ID 取回 Schema 自动反序列化。kafka-examples仓库用一对点击流生成器 会话化消费者把这套机制讲得明明白白。 环境准备一键搭建运行环境运行 Avro 示例需要三个组件全部使用默认配置启动即可ZookeeperKafka BrokerSchema RegistryConfluent 提供默认监听http://localhost:8081克隆仓库并构建生产者git clone https://gitcode.com/gh_mirrors/kaf/kafka-examples cd kafka-examples/AvroProducerExample mvn clean package然后创建主题clicksbin/kafka-topics.sh --create --zookeeper localhost:2181 \ --replication-factor 1 --partitions 1 --topic clicks 依赖版本参考 AvroProducerExample/pom.xmlKafka 2.4.0、Confluent 5.4.0、Avro 1.7.7其中kafka-avro-serializer是核心依赖。 定义 Avro Schema一切从 LogLine.avsc 开始项目的数据模型是一个网页点击日志LogLineSchema 定义在AvroProducerExample/src/main/resources/avro/LogLine.avsc{ namespace: JavaSessionize.avro, type: record, name: LogLine, fields: [ {name: ip, type: string}, {name: timestamp, type: long}, {name: url, type: string}, {name: referrer, type: string}, {name: useragent, type: string}, {name: sessionid, type: [null,int], default: null} ] }注意namespacename组合成了 Schema Registry 中的全限定主题名JavaSessionize.avro.LogLine这是 Schema 注册与查找的关键标识。构建时avro-maven-plugin会在generate-sources阶段基于该 Schema自动生成 Java 类Specific 类代码里直接new LogLine()填充字段即可无需手写序列化逻辑。 第一步生产者——用 KafkaAvroSerializer 发送 Avro 事件核心代码在AvroProducerExample/src/main/java/com/shapira/examples/producer/avroclicks/AvroClicksProducer.java配置 Producer 只需关注 4 个参数props.put(key.serializer, io.confluent.kafka.serializers.KafkaAvroSerializer); props.put(value.serializer, io.confluent.kafka.serializers.KafkaAvroSerializer); props.put(schema.registry.url, schemaUrl); props.put(acks, all);EventGenerator同目录EventGenerator.java负责生成模拟点击事件生产者的发送逻辑非常直观LogLine event EventGenerator.getNext(); ProducerRecordString, LogLine record new ProducerRecord(clicks, event.getIp().toString(), event); producer.send(record).get();幕后发生了什么KafkaAvroSerializer首次序列化LogLine时向 Schema Registry 注册其 Schema 并拿到 ID消息体被编码为1 字节魔数 4 字节 Schema ID Avro 二进制数据同一 IP 的事件因为 Key 相同会被路由到同一分区——天然保证了同一用户的事件有序运行生产者写入 100 条点击java -cp target/uber-ClickstreamGenerator-1.0-SNAPSHOT.jar \ com.shapira.examples.producer.avroclicks.AvroClicksProducer 100 http://localhost:8081 第二步消费者——KafkaAvroDeserializer 自动反序列化消费端示例在AvroConsumerExample/src/main/java/com/shapira/examples/consumer/avroclicks/AvroClicksSessionizer.java配置与生产端对称props.put(key.deserializer, org.apache.kafka.common.serialization.StringDeserializer); props.put(value.deserializer, io.confluent.kafka.serializers.KafkaAvroDeserializer); props.put(schema.registry.url, url); props.put(specific.avro.reader, true);关键配置速查表配置项作用示例值key.serializer/value.serializer生产端序列化器KafkaAvroSerializerkey.deserializer/value.deserializer消费端反序列化器KafkaAvroDeserializerschema.registry.urlSchema Registry 地址http://localhost:8081specific.avro.reader用生成的 Specific 类读取trueauto.offset.reset无消费位点时的起点earliestspecific.avro.reader true表示消费时直接还原为编译期生成的LogLine类而不是通用的GenericRecord——类型安全且无需手动取字段。该消费者读取clicks主题后做了一件典型的事会话化。它用内存表记录每个 IP 的最后活跃时间间隔超过 30 分钟就递增sessionid状态管理见同目录SessionState.java再把带会话 ID 的事件写入sessionized_clicks主题——正是消费 → 加工 → 再生产的经典管道模式。运行前先创建输出主题bin/kafka-topics.sh --create --zookeeper localhost:2181 \ --replication-factor 1 --partitions 1 --topic sessions⚠️ 新手避坑清单Schema Registry 必须先启动生产者启动时会连接它注册 Schema连不上会直接抛异常namespace别乱改它决定 Schema 在 Registry 中的全名生产与消费两端必须一致Key 的序列化器是独立的本例 Key 是 IP 字符串用StringSerializer即可无需套用 Avro手动提交位点消费端关闭了自动提交auto.commit.enablefalse处理完一批后commitSync()保证不丢消息验证结果用 Confluent 自带的 Avro 控制台消费者查看落库数据bin/kafka-avro-console-consumer --zookeeper localhost:2181 --topic sessionized_clicks --from-beginning 总结通过 kafka-examples 的 Avro 生产/消费示例你掌握了完整的 Avro 序列化链路用LogLine.avsc定义 SchemaMaven 插件自动生成 Java 类生产端配置KafkaAvroSerializerschema.registry.url完成注册与编码消费端配置KafkaAvroDeserializerspecific.avro.reader自动还原强类型对象借助 Schema Registry 实现 Schema 的集中管理与版本演进这套生产 消费 Schema 治理的组合拳正是企业级 Kafka 数据管道的标准姿势。建议继续阅读仓库中的 AvroProducerExample/README.md 和 AvroConsumerExample/README.md动手跑通后再挑战 Kafka Streams 相关示例进阶流式计算 【免费下载链接】kafka-examplesSnippets and small examples demonstrating kafka features and configs项目地址: https://gitcode.com/gh_mirrors/kaf/kafka-examples创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
