你当前正在访问 Microsoft Azure Global Edition 技术文档网站。如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站，请访问 https://docs.azure.cn。

将 JSON 架构与 Apache Kafka 应用程序配合使用

项目
10/21/2023

本教程将引导你完成使用 JSON 架构在事件中心通过 Azure 架构注册表序列化和反序列化事件的方案。

在此用例中，Kafka 生成者应用程序会使用存储在 Azure 架构注册表中的 JSON 架构来序列化事件并将其发布到 Azure 事件中心中的 Kafka 主题/事件中心。 Kafka 使用者从事件中心反序列化它使用的事件。为此，它使用存储在 Azure 架构注册表中的事件的架构 ID 和 JSON 架构。显示使用 JSON 架构对 Kafka 应用程序进行序列化/反序列化的关系图。

先决条件

如果不熟悉 Azure 事件中心，请在阅读本快速入门之前参阅事件中心概述。

若要完成本快速入门，需要具备以下先决条件：

如果还没有 Azure 订阅，可以在开始前创建一个免费帐户。
在开发环境中，安装以下组件：
- Java 开发工具包 (JDK) 1.7+。
- 下载并安装 Maven 二进制存档。
- Git
克隆适用于 Kafka 的 Azure 架构注册表存储库。

创建事件中心

按照快速入门创建事件中心命名空间和事件中心中的说明创建事件中心命名空间和事件中心。然后，按照获取连接字符串中的说明获取事件中心命名空间的连接字符串。

记下你在当前快速入门中所使用的以下设置：

事件中心命名空间的连接字符串
事件中心的名称

创建架构

按照使用架构注册表创建架构中的说明创建架构组和架构。

使用架构注册表门户创建名为 contoso-sg 的架构组。使用 JSON 架构 作为序列化类型。

在该架构组中，使用以下架构内容创建名为 Microsoft.Azure.Data.SchemaRegistry.example.CustomerInvoice 的新 JSON 架构。

{
  "$id": "https://example.com/person.schema.json",
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "title": "CustomerInvoice",
  "type": "object",
  "properties": {
    "invoiceId": {
      "type": "string"
    },
    "merchantId": {
      "type": "string"
    },
    "transactionValueUsd": {
      "type": "integer"
    },
    "userId": {
      "type": "string"
    }
  }
}

注册应用程序以访问架构注册表

可以使用 Microsoft Entra ID 授权 Kafka 生成者和使用者应用程序访问 Azure 架构注册表资源。若要启用它，需要从 Azure 门户向 Microsoft Entra 租户注册客户端应用程序。

若要注册名为 example-app 的 Microsoft Entra 应用程序，请参阅向 Microsoft Entra 租户注册应用程序。

tenant.id - 设置应用程序的租户 ID
client.id - 设置应用程序的客户端 ID
client.secret - 设置用于身份验证的客户端密码

如果你在使用托管标识，则需要：

use.managed.identity.credential - 指示应使用 MSI 凭据，应将 MSI 凭据用于已启用 MSI 的 VM
managed.identity.clientId - 如果指定，它将生成具有给定客户端 ID managed.identity.resourceId 的 MSI 凭据 - 如果指定，它将生成具有给定资源 ID 的 MSI 凭据

将用户添加到架构注册表读取者角色

在命名空间级别，将用户帐户添加到“架构注册表读取者”角色。也可以使用“架构注册表参与者”角色，但本快速入门不需要这样做。

在“事件中心命名空间”页面上，从左侧菜单中选择“访问控制(IAM)”。
在“访问控制(IAM)”页面上，依次选择“+ 添加”-> 菜单上的“添加角色分配”。
在“分配类型”页面上，选择“下一步”。
在“角色”页面上，选择“架构注册表读取者”，然后选择页面底部的“下一步”。
使用“+ 选择成员”链接将上一步中创建的 example-app 应用程序添加到角色，然后选择“下一步”。
在“查看 + 分配”页面上，选择“查看 + 分配”。

更新 Kafka 应用程序的客户端应用程序配置

需要使用 Microsoft Entra 应用程序详细信息和架构注册表信息更新 Kafka 生成者和使用者应用程序的客户端配置。

要更新 Kafka 生成者配置，可导航到 azure-schema-registry-for-kafka/tree/master/java/json/samples/kafka-producer。

按照事件中心 Kafka 快速入门指南，在 src/main/resources/app.properties 中更新 Kafka 应用程序的配置。
使用你在之前的步骤中创建的架构注册表相关配置和 Microsoft Entra 应用程序更新 src/main/resources/app.properties 中生成者的配置详细信息，如下所示：
```
schema.group=contoso-sg
schema.registry.url=https://<NAMESPACENAME>.servicebus.windows.net

 tenant.id=<>
 client.id=<>
 client.secret=<>
```
按照相同的说明操作，并更新 azure-schema-registry-for-kafka/tree/master/java/json/samples/kafka-consumer 配置。

对于 Kafka 生成者和使用者应用程序，使用以下 JSON 架构：

{
  "$id": "https://example.com/person.schema.json",
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "title": "CustomerInvoice",
  "type": "object",
  "properties": {
    "invoiceId": {
      "type": "string"
    },
    "merchantId": {
      "type": "string"
    },
    "transactionValueUsd": {
      "type": "integer"
    },
    "userId": {
      "type": "string"
    }
  }
}

将 Kafka 生成者与 JSON 架构验证配合使用

要运行 Kafka 生成者应用程序，可导航到 azure-schema-registry-for-kafka/tree/master/java/json/samples/kafka-consumer。

可以运行生成者应用程序，以便它可以生成特定于 JSON 架构的记录或通用记录。对于特定记录模式，需要首先使用以下 maven 命令针对生成者架构生成类：
```
mvn generate-sources
```

然后，可以使用以下命令运行生成者应用程序。

mvn clean package
mvn -e clean compile exec:java -Dexec.mainClass="com.azure.schemaregistry.samples.producer.App"

成功执行生成者应用程序后，它会提示你选择生成者方案。对于本快速入门，可以选择选项“1 - 生成 SpecificRecords”。
```
Enter case number:
1 - produce SpecificRecords
```

成功进行数据序列化和发布后，应在生成者应用程序中看到以下控制台日志：

INFO com.azure.schemaregistry.samples.producer.KafkaJsonSpecificRecord - Sent Order Invoice 0
INFO com.azure.schemaregistry.samples.producer.KafkaJsonSpecificRecord - Sent Order Invoice 1
INFO com.azure.schemaregistry.samples.producer.KafkaJsonSpecificRecord - Sent Order Invoice 2

将 Kafka 使用者与 JSON 架构验证配合使用

要运行 Kafka 使用者应用程序，可导航到 azure-schema-registry-for-kafka/tree/master/java/json/samples/kafka-consumer。

可以运行使用者应用程序，以便它可以使用特定于 JSON 架构的记录或通用记录。对于特定记录模式，需要首先使用以下 maven 命令针对生成者架构生成类：
```
mvn generate-sources
```

然后，可以使用以下命令运行使用者应用程序。

mvn clean package
mvn -e clean compile exec:java -Dexec.mainClass="com.azure.schemaregistry.samples.consumer.App"

成功执行使用者应用程序后，它会提示你选择生成者方案。对于本快速入门，可以选择选项“1 - 使用 SpecificRecords”。
```
Enter case number:
1 - consume SpecificRecords
```

成功使用和反序列化数据后，生成者应用程序中应会显示以下控制台日志：

INFO com.azure.schemaregistry.samples.consumer.KafkaJsonSpecificRecord - Invoice received: {invoiceId=Invoice 0, merchantId=Merchant Id 0, transactionValueUsd=0, userId=User Id 0}
INFO com.azure.schemaregistry.samples.consumer.KafkaJsonSpecificRecord - Invoice received: {invoiceId=Invoice 1, merchantId=Merchant Id 1, transactionValueUsd=1, userId=User Id 1}
INFO com.azure.schemaregistry.samples.consumer.KafkaJsonSpecificRecord - Invoice received: {invoiceId=Invoice 2, merchantId=Merchant Id 2, transactionValueUsd=2, userId=User Id 2}

清理资源

删除事件中心命名空间或删除包含该命名空间的资源组。

通过