文档首页

租户端互联网中间件消息队列（CKafka）最佳实践

Schema Registry 接入 CKafka

最近更新时间: 2024-06-12 15:06:00

无论是使用传统的 Avro API 自定义序列化类与反序列化类，还是使用 Twitter 的 Bijection 类库实现 Avro 的序列化与反序列化，两种方法有相同的缺点：在每条 Kafka 记录里都嵌入了 Schema，从而导致记录的大小成倍增加。但是不管怎样，在读取记录时仍然需要用到整个 Schema，所以要先找到 Schema。 CKafka 提供了数据共用一个 Schema 的方法：将 Schema 中的内容注册到 Confluent Schema Registry，Kafka Producer 和 Kafka Consumer 通过识别 Confluent Schema Registry 中的 schema 内容进行序列化和反序列化。

前提条件

下载 Download JDK 8。
下载 Confluent oss 4.1.1。
已创建实例]。

操作步骤

步骤1. 获取实例接入地址并开启自动创建Topic

登录 CKafka 控制台。
在左侧导航栏选择【实例列表】，单击实例的“ID”，进入实例基本信息页面。
在实例的基本信息页面的【接入方式】模块，可获取实例的接入地址。
在【自动创建 Topic】模块开启自动创建 Topic。

注意：

启动 oss 会创建 schemas 主题，所以实例中需要开启自动创建主题。

步骤2. 准备Confluent配置

修改 oss 配置文件中的 server 地址等信息。配置信息如下：
```
 kafkastore.bootstrap.servers=PLAINTEXT://xxxx
 kafkastore.topic=schemas
 debug=true
```
说明：

bootstrap.servers：接入网络，在实例详情页面【接入方式】模块的网络列复制。

执行如下命令启动 Schema Registry。

 bin/schema-registry-start etc/schema-registry/schema-registry.properties

运行结果如下：

步骤3. 收发消息

现有 schema 文件，其中内容如下：

{
    "type": "record",
    "name": "User",
    "fields": [
        {"name": "id", "type": "int"},
        {"name": "name",  "type": "string"},
        {"name": "age", "type": "int"}
    ]
}

curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
--data '{"schema": "{\"type\": \"record\", \"name\": \"User\", \"fields\": [{\"name\": \"id\", \"type\": \"int\"}, {\"name\": \"name\",  \"type\": \"string\"}, {\"name\": \"age\", \"type\": \"int\"}]}"}' \
http://127.0.0.1:8081/subjects/test/versions

Kafka Producer 发送数据：

package schemaTest;
import java.util.Properties;
import java.util.Random;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
public class SchemaProduce {
    public static final String USER_SCHEMA = "{\"type\": \"record\", \"name\": \"User\", " + 
               "\"fields\": [{\"name\": \"id\", \"type\": \"int\"}, " + 
               "{\"name\": \"name\",  \"type\": \"string\"}, {\"name\": \"age\", \"type\": \"int\"}]}";
       public static void main(String[] args) throws Exception {
           Properties props = new Properties();
           // 添加CKafka实例的接入地址  
         props.put("bootstrap.servers", "xx.xx.xx.xx:xxxx");
           props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
           // 使用 Confluent 实现的 KafkaAvroSerializer
           props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
           // 添加 schema 服务的地址，用于获取 schema
           props.put("schema.registry.url", "http://127.0.0.1:8081");
           Producer<String, GenericRecord> producer = new KafkaProducer<>(props);           
           Schema.Parser parser = new Schema.Parser();
           Schema schema = parser.parse(USER_SCHEMA);     
           Random rand = new Random();
           int id = 0;
           while(id < 100) {
               id++;
               String name = "name" + id;
               int age = rand.nextInt(40) + 1;
               GenericRecord user = new GenericData.Record(schema);
               user.put("id", id);
               user.put("name", name);
               user.put("age", age);                
               ProducerRecord<String, GenericRecord> record = new ProducerRecord<>("test", user);                
               producer.send(record);
               Thread.sleep(1000);
           }
           producer.close();
       }
}

运行一段时间后，在CKafka 控制台的【topic管理】页面，选择对应的 Topic ，单击【更多】>【消息查询】，查看刚刚发送的消息。

Kafka Consumer 消费数据：

package schemaTest;
import java.util.Collections;
import java.util.Properties;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
public class SchemaProduce {
   public static void main(String[] args) throws Exception {
       Properties props = new Properties();
       props.put("bootstrap.servers", "xx.xx.xx.xx:xxxx"); //CKafka实例的接入地址
       props.put("group.id", "schema");
       props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
       // 使用Confluent实现的KafkaAvroDeserializer
       props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
       // 添加schema服务的地址，用于获取schema
       props.put("schema.registry.url", "http://127.0.0.1:8081");
       KafkaConsumer<String, GenericRecord> consumer = new KafkaConsumer<>(props);
       consumer.subscribe(Collections.singletonList("test"));
       try {
           while (true) {
               ConsumerRecords<String, GenericRecord> records = consumer.poll(10);
               for (ConsumerRecord<String, GenericRecord> record : records) {
                   GenericRecord user = record.value();
                   System.out.println("value = [user.id = " + user.get("id") + ", " + "user.name = "
                           + user.get("name") + ", " + "user.age = " + user.get("age") + "], "
                           + "partition = " + record.partition() + ", " + "offset = " + record.offset());
               }
           }
       } finally {
           consumer.close();
       }
   }
}

在 CKafka 控制台的【Consumer Group】页面，选择 schema消费组名称，在主题名称输入 Topic 名称，单击【查询详情】，查看消费详情。

启动消费者进行消费，下图为消费日志截图：