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

教程：连接到 Azure 事件网格的双向 MQTT 网桥

项目
01/08/2025

重要

本页包含使用 Kubernetes 部署清单（目前为预览版）管理 Azure IoT 操作组件的说明。此功能存在若干限制，不应该用于生产工作负载。

有关 beta 版本、预览版或尚未正式发布的版本的 Azure 功能所适用的法律条款，请参阅 Microsoft Azure 预览版的补充使用条款。

在本教程中，我们在 Azure IoT 操作 MQTT 代理和 Azure 事件网格之间设置双向 MQTT 网桥。为了简化教程，请使用 Azure IoT 操作 MQTT 代理和 Azure 事件网格终结点的默认设置，并且不应用任何转换。

先决条件

Azure IoT 操作。请参阅部署 Azure IoT 操作。
数据流配置文件。请参阅配置数据流配置文件。

设置环境变量。

使用 Azure CLI 登录：

az login

设置其余设置的环境变量。将<>中的值替换为所选的有效值或名称。会根据你提供的名称在 Azure 订阅中创建新的 Azure 事件网格命名空间和主题空间：

Bash
PowerShell

# For this tutorial, the steps assume the IoT Operations cluster and the Event Grid
# are in the same subscription, resource group, and location.

# Name of the resource group of Azure Event Grid and IoT Operations cluster 
export RESOURCE_GROUP=<RESOURCE_GROUP_NAME>

# Azure region of Azure Event Grid and IoT Operations cluster
export LOCATION=<LOCATION>

# Name of the Azure Event Grid namespace
export EVENT_GRID_NAMESPACE=<EVENT_GRID_NAMESPACE>

# Name of the Arc-enabled IoT Operations cluster 
export CLUSTER_NAME=<CLUSTER_NAME>

# Subscription ID of Azure Event Grid and IoT Operations cluster
export SUBSCRIPTION_ID=<SUBSCRIPTION_ID>

# For this tutorial, the steps assume the IoT Operations cluster and the Event Grid
# are in the same subscription, resource group, and location.

# Name of the resource group of Azure Event Grid and IoT Operations cluster 
$RESOURCE_GROUP = "<RESOURCE_GROUP_NAME>"

# Azure region of Azure Event Grid and IoT Operations cluster
$LOCATION = "<LOCATION>"

# Name of the Azure Event Grid namespace
$EVENT_GRID_NAMESPACE = "<EVENT_GRID_NAMESPACE>"

# Name of the Arc-enabled IoT Operations cluster 
$CLUSTER_NAME = "<CLUSTER_NAME>"

# Subscription ID of Azure Event Grid and IoT Operations cluster
$SUBSCRIPTION_ID = "<SUBSCRIPTION_ID>"

创建启用了 MQTT 代理的 Azure 事件网格命名空间

使用 Azure CLI 创建事件网格命名空间。位置应与用于部署 Azure IoT 操作的位置相同。

Bash
PowerShell

az eventgrid namespace create \
  --namespace-name $EVENT_GRID_NAMESPACE \
  --resource-group $RESOURCE_GROUP \
  --location $LOCATION \
  --topic-spaces-configuration "{state:Enabled,maximumClientSessionsPerAuthenticationName:3}"

az eventgrid namespace create `
  --namespace-name $EVENT_GRID_NAMESPACE `
  --resource-group $RESOURCE_GROUP `
  --location $LOCATION `
  --topic-spaces-configuration "{state:Enabled,maximumClientSessionsPerAuthenticationName:3}"

如果未安装 eventgrid 扩展，你会收到一个询问你是否要安装它的提示。选择“Y”来安装该扩展。

通过设置 topic-spaces-configuration，此命令会创建一个命名空间：

已启用 MQTT 代理
每个身份验证名称的最大客户端会话数为 3。

最大客户端会话选项允许 Azure IoT 操作 MQTT 生成多个实例，并且仍可连接。若要了解详细信息，请参阅多会话支持。

创建主题空间

在 Azure 事件网格命名空间中，使用主题模板 telemetry/# 创建名为 tutorial 的主题空间。

Bash
PowerShell

az eventgrid namespace topic-space create \
  --resource-group $RESOURCE_GROUP \
  --namespace-name $EVENT_GRID_NAMESPACE \
  --name tutorial \
  --topic-templates "telemetry/#"

az eventgrid namespace topic-space create `
  --resource-group $RESOURCE_GROUP `
  --namespace-name $EVENT_GRID_NAMESPACE `
  --name tutorial `
  --topic-templates "telemetry/#"

通过在主题模板中使用 # 通配符，可以发布到 telemetry 主题空间下的任何主题。例如，telemetry/temperature 或 telemetry/humidity。

授予 Azure IoT 操作对事件网格主题空间的访问权限

使用 Azure CLI，查找 Azure IoT 操作 Arc 扩展的主体 ID。该命令将主体 ID 存储在变量中供以后使用。

Bash
PowerShell

export PRINCIPAL_ID=$(az k8s-extension list \
  --resource-group $RESOURCE_GROUP \
  --cluster-name $CLUSTER_NAME \
  --cluster-type connectedClusters \
  --query "[?extensionType=='microsoft.iotoperations'].identity.principalId | [0]" -o tsv)
echo $PRINCIPAL_ID

$PRINCIPAL_ID = (az k8s-extension list `
  --resource-group $RESOURCE_GROUP `
  --cluster-name $CLUSTER_NAME `
  --cluster-type connectedClusters `
  --query "[?extensionType=='microsoft.iotoperations'].identity.principalId | [0]" -o tsv)
Write-Output $PRINCIPAL_ID

记下 identity.principalId 的输出值，它是采用以下格式的 GUID 值：

aaaaaaaa-bbbb-cccc-1111-222222222222

然后，使用 Azure CLI 为创建的主题空间将发布服务器和订阅服务器角色分配给 Azure IoT 操作 MQTT。

分配发布者角色：

Bash
PowerShell

az role assignment create \
  --assignee $PRINCIPAL_ID \
  --role "EventGrid TopicSpaces Publisher" \
  --scope /subscriptions/$SUBSCRIPTION_ID/resourceGroups/$RESOURCE_GROUP/providers/Microsoft.EventGrid/namespaces/$EVENT_GRID_NAMESPACE/topicSpaces/tutorial

az role assignment create `
  --assignee $PRINCIPAL_ID `
  --role "EventGrid TopicSpaces Publisher" `
  --scope /subscriptions/$SUBSCRIPTION_ID/resourceGroups/$RESOURCE_GROUP/providers/Microsoft.EventGrid/namespaces/$EVENT_GRID_NAMESPACE/topicSpaces/tutorial

分配订阅者角色：

Bash
PowerShell

az role assignment create \
  --assignee $PRINCIPAL_ID \
  --role "EventGrid TopicSpaces Subscriber" \
  --scope /subscriptions/$SUBSCRIPTION_ID/resourceGroups/$RESOURCE_GROUP/providers/Microsoft.EventGrid/namespaces/$EVENT_GRID_NAMESPACE/topicSpaces/tutorial

az role assignment create `
  --assignee $PRINCIPAL_ID `
  --role "EventGrid TopicSpaces Subscriber" `
  --scope /subscriptions/$SUBSCRIPTION_ID/resourceGroups/$RESOURCE_GROUP/providers/Microsoft.EventGrid/namespaces/$EVENT_GRID_NAMESPACE/topicSpaces/tutorial

提示

该范围与在上一步中使用 az eventgrid namespace topic-space create 创建的主题空间的 id 相匹配，可以在命令的输出中找到它。

事件网格 MQTT 代理主机名

使用 Azure CLI 获取事件网格 MQTT 代理主机名。

Bash
PowerShell

az eventgrid namespace show \
  --resource-group $RESOURCE_GROUP \
  --namespace-name $EVENT_GRID_NAMESPACE \
  --query topicSpacesConfiguration.hostname \
  -o tsv

az eventgrid namespace show `
  --resource-group $RESOURCE_GROUP `
  --namespace-name $EVENT_GRID_NAMESPACE `
  --query topicSpacesConfiguration.hostname `
  -o tsv

记下 topicSpacesConfiguration.hostname 的输出值，其主机名值如下所示：

example.region-1.ts.eventgrid.azure.net

创建 Azure 事件网格数据流终结点

为 Azure 事件网格创建数据流终结点。此终结点是向 Azure 事件网格发送消息的数据流的目标。将 <EVENT_GRID_HOSTNAME> 替换为从上一步获取的 MQTT 主机名。包括端口号 8883。

Bicep
Kubernetes（预览版）

数据流和数据流终结点 Azure 事件网格可以部署为标准 Azure 资源，因为它们具有 Azure 资源提供程序 (RP) 实现。此 Bicep 模板文件来自 MQTT 网桥数据流教程的 Bicep 文件，可部署必要的数据流和数据流终结点。

将文件下载到本地，并确保将 customLocationName、aioInstanceName、eventGridHostName 替换为你自己的值。

param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param eventGridHostName string = '<EVENT_GRID_HOSTNAME>:8883'

resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
  name: customLocationName
}

resource aioInstance 'Microsoft.IoTOperations/instances@2024-11-01' existing = {
  name: aioInstanceName
}
resource remoteMqttBrokerDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-11-01' = {
  parent: aioInstance
  name: 'eventgrid'
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    endpointType: 'Mqtt'
    mqttSettings: {
      host: eventGridHostName
      authentication: {
        method: 'SystemAssignedManagedIdentity'
        systemAssignedManagedIdentitySettings: {}
      }
      tls: {
        mode: 'Enabled'
      }
    }
  }
}

接下来，在终端中执行以下命令。将 <FILE> 替换为你已下载的 Bicep 文件的名称。

az deployment group create --resource-group <RESOURCE_GROUP> --template-file <FILE>.bicep

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowEndpoint
metadata:
  name: eventgrid
  namespace: azure-iot-operations
spec:
  endpointType: Mqtt
  mqttSettings:
    host: <EVENT_GRID_HOSTNAME>:8883
    authentication:
      method: SystemAssignedManagedIdentity
      systemAssignedManagedIdentitySettings: {}
    tls:
      mode: Enabled

在这里，身份验证方法设置为 SystemAssignedManagedIdentity，以使用 Azure IoT 操作扩展的托管标识通过事件网格 MQTT 代理进行身份验证。此设置有效，因为 Azure IoT 操作扩展具有发布和订阅通过 Azure RBAC 角色配置的事件网格主题空间所需的权限。请注意，配置中不需要用户名或密码之类的机密。

由于事件网格 MQTT 代理需要 TLS，因此启用了 tls 设置。无需提供受信任的 CA 证书，因为事件网格 MQTT 代理使用广受信任的证书颁发机构。

创建数据流

创建两个数据流，以 Azure IoT 操作 MQTT 代理终结点为源，以 Azure 事件网格终结点为目标，反之亦可。无需配置转换。

Bicep
Kubernetes（预览版）

param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param aioInstanceName string = '<AIO_INSTANCE_NAME>'

resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
  name: customLocationName
}
resource aioInstance 'Microsoft.IoTOperations/instances@2024-11-01' existing = {
  name: aioInstanceName
}
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-11-01' existing = {
  parent: aioInstance
  name: 'default'
}
resource dataflow_1 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-11-01' = {
  parent: defaultDataflowProfile
  name: 'local-to-remote'
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    operations: [
      {
        operationType: 'Source'
        sourceSettings: {
          endpointRef: 'default'
          serializationFormat: 'Json'
          dataSources: array('tutorial/local')
        }
      }
      {
        operationType: 'BuiltInTransformation'

        builtInTransformationSettings: {
        serializationFormat: 'Json'
        datasets: []
        filter: []
        map: [
          {
            type: 'PassThrough'
            inputs: [
              '*'
            ]
            output: '*'
          }
        ]
        }
      }
      {
        operationType: 'Destination'
        destinationSettings: {
          endpointRef: 'eventgrid'
          dataDestination: 'telemetry/aio'
        }
      }
    ]
  }
} 
resource dataflow_2 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-11-01' = {
  parent: defaultDataflowProfile
  name: 'remote-to-local'
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    operations: [
      {
        operationType: 'Source'
        sourceSettings: {
          endpointRef: 'eventgrid'
          serializationFormat: 'Json'
          dataSources: array('telemetry/#')
        }
      }
      {
        operationType: 'BuiltInTransformation'

        builtInTransformationSettings: {
        serializationFormat: 'Json'
        datasets: []
        filter: []
        map: [
          {
            type: 'PassThrough'
            inputs: [
              '*'
            ]
            output: '*'
          }
        ]
        }
      }
      {
        operationType: 'Destination'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'tutorial/cloud'
        }
      }
    ]
  }
}

与数据流终结点一样，请在终端中执行以下命令：

az deployment group create --resource-group <RESOURCE_GROUP> --template-file <FILE>.bicep

apiVersion: connectivity.iotoperations.azure.com/v1
kind: Dataflow
metadata:
  name: local-to-remote
  namespace: azure-iot-operations
spec:
  profileRef: default
  operations:
  - operationType: Source
    sourceSettings:
      endpointRef: default
      dataSources:
        - tutorial/local
  - operationType: Destination
    destinationSettings:
      endpointRef: eventgrid
      dataDestination: telemetry/aio
---
apiVersion: connectivity.iotoperations.azure.com/v1
kind: Dataflow
metadata:
  name: remote-to-local
  namespace: azure-iot-operations
spec:
  profileRef: default
  operations:
  - operationType: Source
    sourceSettings:
      endpointRef: eventgrid
      dataSources:
        - telemetry/#
  - operationType: Destination
    destinationSettings:
      endpointRef: default
      dataDestination: tutorial/cloud

这两个数据流共同构成了 MQTT 网桥，你可以在其中执行以下操作：

使用 Azure 事件网格 MQTT 代理作为远程代理
使用本地 Azure IoT 操作 MQTT 代理作为本地代理
对远程和本地代理使用 TLS
使用系统分配的托管标识向远程代理进行身份验证
使用 Kubernetes 服务帐户向本地代理进行身份验证
使用主题映射将 tutorial/local 主题映射到远程代理上的 telemetry/aio 主题
使用主题映射将远程代理上的 telemetry/# 主题映射到本地代理上的 tutorial/cloud 主题

注意

默认情况下，Azure IoT 操作部署 MQTT 代理以及 MQTT 代理数据流终结点。 MQTT 代理数据流终结点用于连接到 MQTT 代理。默认配置使用内置服务帐户令牌进行身份验证。终结点名为 default，在与 Azure IoT 操作相同的命名空间中可用。终结点用作在本教程中创建的数据流的源。若要详细了解默认 MQTT 代理数据流终结点，请参阅 Azure IoT 操作本地 MQTT 代理默认终结点。

发布到本地 Azure IoT 操作 MQTT 代理上的 tutorial/local 主题时，消息会桥接至远程事件网格 MQTT 代理上的 telemetry/aio 主题。然后，消息会桥接回本地 Azure IoT 操作 MQTT 代理上的 tutorial/cloud 主题（因为 telemetry/# 通配符主题会捕获它）。同样，在远程 Azure 事件网格 MQTT 代理上发布到 telemetry/aio 主题时，消息会桥接到本地 Azure IoT 操作 MQTT 代理上的 tutorial/cloud 主题。

部署 MQTT 客户端

若要验证 MQTT 网桥是否正常工作，请将 MQTT 客户端部署到与 Azure IoT 操作相同的命名空间。

Bicep
Kubernetes（预览版）

目前，Bicep 不适用于部署 MQTT 客户端。

从 GitHub 示例存储库下载 mqtt-client.yaml 部署。

重要

请勿在生产环境中使用 MQTT 客户端。该客户端仅用于测试目的。

wget https://raw.githubusercontent.com/Azure-Samples/explore-iot-operations/main/samples/quickstarts/mqtt-client.yaml -O mqtt-client.yaml

使用 kubectl 应用部署文件。

kubectl apply -f mqtt-client.yaml

pod/mqtt-client created

启动订阅服务器

使用 kubectl exec 在 mosquitto 客户端 Pod 中启动 shell。

kubectl exec --stdin --tty mqtt-client -n azure-iot-operations -- sh

在 shell 中，使用 mosquitto_sub 在 tutorial/# 主题空间上启动 Azure IoT 操作代理的订阅服务器。

mosquitto_sub --host aio-broker --port 18883 \
  -t "tutorial/#" \
  --debug --cafile /var/run/certs/ca.crt \
  -D CONNECT authentication-method 'K8S-SAT' \
  -D CONNECT authentication-data $(cat /var/run/secrets/tokens/broker-sat)

使命令保持运行状态，并打开新的终端窗口。

通过桥将 MQTT 消息发布到云

在新的终端窗口中，在 mosquitto 客户端 Pod 中启动另一个 shell。

kubectl exec --stdin --tty mqtt-client -n azure-iot-operations -- sh

在 shell 中，使用 mosquitto 将五条消息发布到 tutorial/local 主题。

mosquitto_pub -h aio-broker -p 18883 \
  -m "This message goes all the way to the cloud and back!" \
  -t "tutorial/local" \
  --repeat 5 --repeat-delay 1 -d \
  --debug --cafile /var/run/certs/ca.crt \
  -D CONNECT authentication-method 'K8S-SAT' \
  -D CONNECT authentication-data $(cat /var/run/secrets/tokens/broker-sat)

查看订阅服务器中的消息

在订阅服务器 shell 中，可看到已发布的消息。

Client null sending CONNECT
Client $server-generated/0000aaaa-11bb-cccc-dd22-eeeeee333333 received CONNACK (0)
Client $server-generated/0000aaaa-11bb-cccc-dd22-eeeeee333333 sending SUBSCRIBE (Mid: 1, Topic: tutorial/#, QoS: 0, Options: 0x00)
Client $server-generated/0000aaaa-11bb-cccc-dd22-eeeeee333333 received SUBACK
Subscribed (mid: 1): 0
Client $server-generated/0000aaaa-11bb-cccc-dd22-eeeeee333333 sending PINGREQ
Client $server-generated/0000aaaa-11bb-cccc-dd22-eeeeee333333 received PINGRESP
Client $server-generated/0000aaaa-11bb-cccc-dd22-eeeeee333333 received PUBLISH (d0, q0, r0, m0, 'tutorial/local', ... (52 bytes))
This message goes all the way to the cloud and back!
Client $server-generated/0000aaaa-11bb-cccc-dd22-eeeeee333333 received PUBLISH (d0, q0, r0, m0, 'tutorial/local', ... (52 bytes))
This message goes all the way to the cloud and back!
Client $server-generated/0000aaaa-11bb-cccc-dd22-eeeeee333333 received PUBLISH (d0, q0, r0, m0, 'tutorial/local', ... (52 bytes))
This message goes all the way to the cloud and back!
Client $server-generated/0000aaaa-11bb-cccc-dd22-eeeeee333333 received PUBLISH (d0, q0, r0, m0, 'tutorial/local', ... (52 bytes))
This message goes all the way to the cloud and back!
Client $server-generated/0000aaaa-11bb-cccc-dd22-eeeeee333333 received PUBLISH (d0, q0, r0, m0, 'tutorial/local', ... (52 bytes))
This message goes all the way to the cloud and back!
Client $server-generated/0000aaaa-11bb-cccc-dd22-eeeeee333333 sending PINGREQ
Client $server-generated/0000aaaa-11bb-cccc-dd22-eeeeee333333 received PINGRESP

在这里，你将看到消息发布到本地 Azure IoT 操作代理并发布到 tutorial/local 主题，桥接到 Azure 事件网格 MQTT 代理，然后在 tutorial/cloud 主题上再次桥接回本地 Azure IoT 操作代理。然后将消息传递到订阅服务器。在此示例中，往返时间约为 80 毫秒。

检查 Azure 事件网格指标以验证消息传递

还可以检查 Azure 事件网格指标，以验证消息是否已传递到 Azure 事件网格 MQTT 代理。在 Microsoft Azure 门户中，转到创建的事件网格命名空间。在“指标”>“MQTT: 成功发布的消息”下。在将消息发布到本地 Azure IoT 操作代理时，应会看到已发布和传递的消息数增加。

Microsoft Azure 门户中指标视图的屏幕截图，显示成功的 MQTT 消息。

提示

可以使用 CLI 扩展az iot ops check --detail-level 2 检查数据流、QoS 和消息路由的配置。

后续步骤

本教程介绍了如何使用 Azure 事件网格 MQTT 代理为双向 MQTT 网桥配置 Azure IoT 操作。后续步骤将探讨以下方案：

若要使用 MQTT 客户端将消息直接发布到 Azure 事件网格 MQTT 代理，请参阅将 MQTT 消息发布到事件网格 MQTT 代理。为客户端提供与你创建的主题空间的发布服务器权限绑定，可以将消息发布到 telemetry 下的任何主题，如 telemetry/temperature 或 telemetry/humidity。所有这些消息都桥接到本地 Azure IoT 操作代理上的 tutorial/cloud 主题。
若要为 Azure 事件网格 MQTT 代理设置传递规则，请参阅配置事件网格 MQTT 代理的传递规则。可以使用传递规则根据主题名称将消息路由到不同的主题，或基于消息内容筛选消息。

通过

教程：连接到 Azure 事件网格的双向 MQTT 网桥

先决条件

设置环境变量。

创建启用了 MQTT 代理的 Azure 事件网格命名空间

创建主题空间

授予 Azure IoT 操作对事件网格主题空间的访问权限

事件网格 MQTT 代理主机名

创建 Azure 事件网格数据流终结点

创建数据流

部署 MQTT 客户端

启动订阅服务器

通过桥将 MQTT 消息发布到云

查看订阅服务器中的消息

检查 Azure 事件网格指标以验证消息传递

后续步骤

反馈

其他资源

通过

教程：连接到 Azure 事件网格的双向 MQTT 网桥

先决条件

设置环境变量。

创建启用了 MQTT 代理的 Azure 事件网格命名空间

创建主题空间

授予 Azure IoT 操作对事件网格主题空间的访问权限

事件网格 MQTT 代理主机名

创建 Azure 事件网格数据流终结点

创建数据流

部署 MQTT 客户端

启动订阅服务器

通过桥将 MQTT 消息发布到云

查看订阅服务器中的消息

检查 Azure 事件网格指标以验证消息传递

后续步骤

相关内容

反馈

其他资源