kafka之protobuf

Protobuf 的 .proto 文件是一种描述消息结构的定义文件，使用这种文件可以定义数据结构（消息），然后生成对应语言的类或代码用于序列化和反序列化数据。生成 .proto 文件涉及到编写 .proto 文件定义，然后通过 protoc 编译器生成目标语言的代码（如 Java、Python、Go 等）。

生成 .proto 文件的步骤

1. 编写 .proto 文件

首先，手动编写 .proto 文件来定义消息的结构。每个 .proto 文件定义了消息类型、字段以及字段的类型和编号。

例如，下面的 .proto 文件定义了一个图片消息的结构，包括文件名、格式和二进制数据：

// image.proto
syntax = "proto3";message ImageRecord {// 文件名string filename = 1;// 文件格式string format = 2;// 二进制数据bytes imageData = 3;
}

2. 使用 protoc 编译 .proto 文件

protoc 是 Google 的 Protocol Buffers 编译器，负责将 .proto 文件编译成对应编程语言的类文件。这些类文件用于序列化和反序列化数据。

2.1. 安装 protoc

• 下载并安装 protoc：

  • Linux/macOS：使用包管理器安装

# macOS
brew install protobuf# Ubuntu
sudo apt-get install -y protobuf-compiler

Windows：从 官方下载页面 获取并安装。

win 解压上述zip包：

执行的文件就在这里：

为了方便使用可以把这个bin目录配置在系统环境变量里，也可以直接进入到这个文件夹里

如果配置环境变量的话，安装完毕之后验证：

验证安装：

protoc --version

2.2. 编译 .proto 文件

通过 protoc 命令来编译 .proto 文件为目标语言代码。下面是几种常见语言的生成方式。

2.2.1. 生成 Java 代码

# 将 image.proto 编译为 Java 类，生成到指定目录
protoc --java_out=./output image.proto

上述的指令直接在win的cmd命令行里即可完成，记得提前建好 output目录

执行完毕之后就会生成一个 .java文件

编译后，会在 ./output 目录下生成相应的 Java 类（如 ImageRecord.java），你可以直接使用这些类进行 Protobuf 的序列化和反序列化。

3. 使用生成的类

编译生成的类会包含以下功能：

• 序列化：将定义的消息对象转换为二进制格式，适合传输或存储。
• 反序列化：将二进制格式的数据解析回消息对象。

例如，使用生成的 Java 类序列化和反序列化 ImageRecord：

import com.example.proto.ImageRecord;  // 假设包名为 com.example.proto
import java.nio.file.Files;
import java.io.File;public class ProtobufExample {public static void main(String[] args) throws Exception {// 构建 ImageRecord 消息对象ImageRecord image = ImageRecord.newBuilder().setFilename("example.jpg").setFormat("jpg").setImageData(ByteString.copyFrom(Files.readAllBytes(new File("example.jpg").toPath()))).build();// 序列化为二进制数据byte[] serializedData = image.toByteArray();// 反序列化为 ImageRecord 对象ImageRecord deserializedImage = ImageRecord.parseFrom(serializedData);System.out.println("Filename: " + deserializedImage.getFilename());}
}

4. 定义 .proto 文件的规则

以下是 .proto 文件的常见语法：

• syntax：定义 Protobuf 版本，推荐使用 proto3，较为简洁并且是最新的标准。

syntax = "proto3";

• 消息（Message）定义：使用 message 关键字定义数据结构。

message ImageRecord {string filename = 1;  // string 类型字段，字段编号为 1string format = 2;    // string 类型字段，字段编号为 2bytes imageData = 3;  // 二进制数据，字段编号为 3
}

字段类型：常见的 Protobuf 字段类型包括：

• int32, int64: 整数
• float, double: 浮点数
• bool: 布尔值
• string: 字符串
• bytes: 二进制数据（如文件、图片、视频）

字段编号：每个字段必须有唯一的编号，编号用于序列化和反序列化。编号必须是正整数，1 到 15 的编号用于最常用的字段，因为它们序列化时占用更少的空间

嵌套消息：可以在消息中定义嵌套的消息类型

message User {string username = 1;Profile profile = 2; // 嵌套消息类型message Profile {string email = 1;int32 age = 2;}
}

总结

1. 编写 .proto 文件：定义消息结构，包括字段类型、名称和编号。
2. 使用 protoc 编译：将 .proto 文件编译为目标语言代码，如 Java、Python、Go 等。
3. 使用生成的类：使用生成的类进行消息的序列化（转换为二进制格式）和反序列化（解析二进制数据）。

kafka和protobuf集成例子：

要将 Protobuf 与 Kafka 集成，我们可以使用 Protobuf 定义的数据结构作为 Kafka 消息体，并通过 Kafka Producer 将序列化的 Protobuf 消息发送到 Kafka。在消费者端，通过 Kafka Consumer 接收消息并反序列化为原始的 Protobuf 对象。

步骤：

1. 编写 .proto 文件：定义消息的结构。
2. 使用 protoc 编译生成类：使用 Protobuf 编译器将 .proto 文件编译为 Java/Python 等语言的类。
3. Kafka Producer 发送 Protobuf 消息：使用生成的类，构造 Protobuf 消息并通过 Kafka Producer 发送。
4. Kafka Consumer 接收并反序列化 Protobuf 消息：在 Kafka Consumer 中接收消息，并反序列化为 Protobuf 对象。

1. 编写 Protobuf .proto 文件

例如，定义一个包含图片信息的 ImageRecord.proto 文件：

syntax = "proto3";message ImageRecord {string filename = 1;string format = 2;bytes imageData = 3;
}

2. 使用 protoc 编译生成 Java 类

假设使用 Java，将 .proto 文件编译为 Java 类：

protoc --java_out=./output ImageRecord.proto

3. Kafka Producer 发送 Protobuf 消息

通过 Kafka Producer 发送 Protobuf 格式的消息：

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import java.util.Properties
import java.nio.file.Files
import java.io.File
import com.example.proto.ImageRecordobject ProtobufKafkaProducer {def main(args: Array[String]): Unit = {// Kafka Producer 配置val props = new Properties()props.put("bootstrap.servers", "localhost:9092")props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer")val producer = new KafkaProducer[String, Array[Byte]](props)// 构建 Protobuf 消息val imageBytes = Files.readAllBytes(new File("/path/to/image.jpg").toPath)val imageRecord = ImageRecord.newBuilder().setFilename("image.jpg").setFormat("jpg").setImageData(com.google.protobuf.ByteString.copyFrom(imageBytes)).build()// 序列化并发送 Protobuf 消息到 Kafkaval record = new ProducerRecord[String, Array[Byte]]("image_topic", "image_key", imageRecord.toByteArray)producer.send(record)producer.close()}
}

4. Kafka Consumer 接收并反序列化 Protobuf 消息

通过 Kafka Consumer 接收 Protobuf 消息并反序列化：

import org.apache.kafka.clients.consumer.{KafkaConsumer, ConsumerRecords}
import java.util.Properties
import com.example.proto.ImageRecordobject ProtobufKafkaConsumer {def main(args: Array[String]): Unit = {// Kafka Consumer 配置val props = new Properties()props.put("bootstrap.servers", "localhost:9092")props.put("group.id", "test")props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer")props.put("auto.offset.reset", "earliest")val consumer = new KafkaConsumer[String, Array[Byte]](props)consumer.subscribe(java.util.Arrays.asList("image_topic"))// 消费并反序列化消息while (true) {val records: ConsumerRecords[String, Array[Byte]] = consumer.poll(100)records.forEach { record =>// 反序列化 Protobuf 消息val imageRecord = ImageRecord.parseFrom(record.value())println(s"Filename: ${imageRecord.getFilename}, Format: ${imageRecord.getFormat}")}}}
}

Kafka 与 Protobuf 集成的优势：

1. 高效序列化：Protobuf 生成的二进制格式非常紧凑，适合大数据量和高吞吐场景。
2. 跨语言支持：Protobuf 支持多种语言，因此 Kafka 与 Protobuf 的集成能轻松跨多语言系统工作。
3. Schema 支持：通过 Protobuf，数据结构的变化可以通过 .proto 文件的模式演进进行管理。