一、前提条件
·下载 Download JDK 8。
·下载 Confluent oss 4.1.1。
·已 创建实例。
二、步骤1：获取实例接入地址并开启自动创建 Topic
1. 登录 CKafka 控制台。
2. 在左侧导航栏选择实例列表，单击实例的“ID”，进入实例基本信息页面。
3. 在实例的基本信息页面的接入方式模块，可获取实例的接入地址。
4. 在自动创建 Topic模块开启自动创建 Topic。
注意：启动 oss 会创建 schemas 主题，所以实例中需要开启自动创建主题。
三、步骤2：准备 Confluent 配置
1. 修改 oss 配置文件中的 server 地址等信息。
PLAINTEXT 接入方式，配置信息如下：
kafkastore.bootstrap.servers=PLAINTEXT://xxxxkafkastore.topic=schemasdebug=true
SASL_PLAINTEXT 接入方式，配置信息如下：
kafkastore.bootstrap.servers=SASL_PLAINTEXT://ckafka-xxxx.ap-xxx.ckafka.tencentcloudmq.com:50004
kafkastore.security.protocol=SASL_PLAINTEXT
kafkastore.sasl.mechanism=PLAIN
kafkastore.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='ckafka-xxxx#xxxx' password='xxxx';kafkastore.topic=schemasdebug=true
注意：bootstrap.servers：接入网络，在 CKafka 控制台 的实例详情页面接入方式模块的网络列复制。
注意：
2. 执行如下命令启动 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"} ]}
1. 注册 schema 到对应 Topic（注册 Topic 名为 test）。
下面的脚本是直接在 Schema Registry 部署的环境中使用 curl 命令调用对应 API 实现注册的一个示例：
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
2. 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 ，单击更多 > 消息查询，查看刚刚发送的消息。
3. 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 名称，单击查看消费者详情，查看消费详情。
启动消费者进行消费，下图为消费日志截图：