快速入门：使用 Azure 事件中心 和 Apache Kafka 流式传输数据

1. 为虚拟机启用系统分配的托管标识。 若要详细了解如何在 VM 上配置托管标识，请参阅使用 Azure 门户在 VM 上配置 Azure 资源托管标识。 Azure 资源的托管标识在 Microsoft Entra ID 中为 Azure 服务提供了一个自动托管标识。 可以使用此标识向支持 Microsoft Entra 身份验证的任何服务进行身份验证，这样就无需在代码中插入凭据了。

   

2. 使用创建的事件中心命名空间的“访问控制”页，将 Azure 事件中心数据所有者角色分配给 VM 的托管标识。 Azure 事件中心支持使用 Microsoft Entra ID 对事件中心资源请求进行授权。 借助 Microsoft Entra ID，可使用 Azure 基于角色的访问控制 (Azure RBAC) 授予对安全主体的访问权限，该安全主体可能是用户或应用程序服务主体。

   1. 在 Azure 门户中，导航到你的事件中心命名空间。 转到左侧导航栏中的“访问控制(IAM)”。

   2. 选择“+ 添加”，然后选择“Add role assignment”。

      

   3. 在“角色”选项卡中，选择“Azure 事件中心数据所有者”，然后选择“下一步”按钮。

      

   4. 在“成员”选项卡中的“将访问权限分配给”部分选择“托管标识”。

   5. 选择“+选择成员”链接。

   6. 在“选择托管标识”页上，执行以下步骤：

      1. 选择包含 VM 的 Azure 订阅。

      2. 对于“托管标识”，请选择“虚拟机”

      3. 选择虚拟机的托管标识。

      4. 选择页面底部的“选择”。

         

   7. 选择“查看 + 分配”。

      

3. 重启 VM 并重新登录到为其配置了托管标识的 VM。

4. 克隆用于 Kafka 的 Azure 事件中心存储库。

5. 导航到 azure-event-hubs-for-kafka/tutorials/oauth/java/managedidentity/consumer。

6. 切换到 src/main/resources/ 文件夹，打开 consumer.config。 将 namespacename 替换为事件中心命名空间的名称。

   bootstrap.servers=NAMESPACENAME.servicebus.chinacloudapi.cn:9093
   security.protocol=SASL_SSL
   sasl.mechanism=OAUTHBEARER
   sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;
   sasl.login.callback.handler.class=CustomAuthenticateCallbackHandler;
   

   注意

   可在此处找到 Kafka 的事件中心的所有 OAuth 示例。

7. 切换回 pom.xml 文件所在的 Consumer 文件夹，使用 Kafka 客户端运行使用者代码并处理来自事件中心的事件：

   mvn clean package
   mvn exec:java -Dexec.mainClass="TestConsumer"                                    
   

8. 启动另一个命令提示符窗口，并导航到 azure-event-hubs-for-kafka/tutorials/oauth/java/managedidentity/producer。

9. 切换到 src/main/resources/ 文件夹，打开 producer.config。 将 mynamespace 替换为事件中心命名空间的名称。

10. 切换回 pom.xml 文件所在的 Producer 文件夹，运行生成者代码并将事件流式传输到事件中心：

    mvn clean package
    mvn exec:java -Dexec.mainClass="TestProducer"                                    
    

    应会看到有关在生成者窗口中发送的事件的消息。 现在请检查使用者应用窗口，了解它从事件中心接收的消息。

    

1. 克隆用于 Kafka 的 Azure 事件中心存储库。

2. 导航到 azure-event-hubs-for-kafka/quickstart/java/consumer。

3. 在 src/main/resources/consumer.config 中更新使用者的配置详细信息，如下所示：

   bootstrap.servers=NAMESPACENAME.servicebus.chinacloudapi.cn:9093
   security.protocol=SASL_SSL
   sasl.mechanism=PLAIN
   sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{YOUR.EVENTHUBS.CONNECTION.STRING}";
   

   重要

   将 {YOUR.EVENTHUBS.CONNECTION.STRING} 替换为事件中心命名空间的连接字符串。 有关获取连接字符串的说明，请参阅获取事件中心连接字符串。 下面是一个配置示例：sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://mynamespace.servicebus.chinacloudapi.cn/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=XXXXXXXXXXXXXXXX";

4. 使用 Kafka 客户端运行使用者代码并处理来自事件中心的事件：

   mvn clean package
   mvn exec:java -Dexec.mainClass="TestConsumer"
   

5. 导航到 azure-event-hubs-for-kafka/quickstart/java/producer。

6. 在 src/main/resources/producer.config 中更新生成者的配置详细信息，如下所示：

   bootstrap.servers=NAMESPACENAME.servicebus.chinacloudapi.cn:9093
   security.protocol=SASL_SSL
   sasl.mechanism=PLAIN
   sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{YOUR.EVENTHUBS.CONNECTION.STRING}";
   

   重要

   将 {YOUR.EVENTHUBS.CONNECTION.STRING} 替换为事件中心命名空间的连接字符串。 有关获取连接字符串的说明，请参阅获取事件中心连接字符串。 下面是一个配置示例：sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://mynamespace.servicebus.chinacloudapi.cn/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=XXXXXXXXXXXXXXXX";

7. 运行生产者代码并将事件流式传输到事件中心：

   mvn clean package
   mvn exec:java -Dexec.mainClass="TestProducer"
   

如果事件中心 Kafka 群集有事件，请现在开始从使用者接收这些事件。