跳到主要内容

mqtt-proxy

mqtt-proxy 插件会校验 MQTT CONNECT 数据包中的协议名称和级别，再将 TCP 连接代理到 MQTT Broker。该插件支持 MQTT 3.1.x 和 MQTT 5.0，并配置在流路由上。

该插件还会将 MQTT 客户端 ID 暴露为 mqtt_client_id。一致性哈希上游可以使用此变量，将相同客户端 ID 的连接保持在同一个 Broker 上。

示例​

以下示例先配置一个流监听器和两个 Eclipse Mosquitto Broker，再代理 MQTT 3.1.1 和 MQTT 5 流量。最后一个示例会添加一致性哈希，使相同客户端 ID 的连接保持在同一个 Broker 上。

配置流监听器​

网关必须在流路由使用的端口上监听流量。

在 config.yaml 中添加流监听器：

config.yaml
apisix:
  proxy_mode: http&stream
  stream_proxy:
    tcp:
      - 9100

保存配置后重新加载网关。

启动示例 MQTT Broker​

这些可运行示例使用 Docker 或 Kubernetes。两种方式都会在网关所在环境中启动两个 Eclipse Mosquitto Broker。

创建 Broker 配置：

mosquitto.conf
listener 1883 0.0.0.0
allow_anonymous true
persistence false
log_dest stdout

将 GATEWAY_CONTAINER 设置为正在运行的 APISIX 或 API7 Gateway 容器：

export GATEWAY_CONTAINER=replace-with-gateway-container-name

创建专用网络并将网关连接到该网络：

docker network create gateway-mqtt-proxy-net
docker network connect gateway-mqtt-proxy-net "$GATEWAY_CONTAINER"

在同一网络中启动两个 Broker。网络别名与下文使用的 Kubernetes Service 地址一致，因此同一套路由配置可用于两种环境：

docker run -d \
  --name mqtt-broker-1 \
  --network gateway-mqtt-proxy-net \
  --network-alias mqtt-broker-1.aic.svc.cluster.local \
  -v "$PWD/mosquitto.conf:/mosquitto/config/mosquitto.conf:ro" \
  eclipse-mosquitto:2.1.2-alpine

docker run -d \
  --name mqtt-broker-2 \
  --network gateway-mqtt-proxy-net \
  --network-alias mqtt-broker-2.aic.svc.cluster.local \
  -v "$PWD/mosquitto.conf:/mosquitto/config/mosquitto.conf:ro" \
  eclipse-mosquitto:2.1.2-alpine

该测试环境允许匿名未加密连接，仅用于评估。生产部署应根据网络拓扑添加适当的身份认证、加密和网络控制。由于 mqtt-proxy 会检查 MQTT CONNECT 数据包，因此无法处理使用 TLS 透传的连接。

代理 MQTT 3.1.1 流量​

本示例在 9100 端口上配置流路由，以接收 MQTT 3.1.1 连接并将其代理到已配置的 Broker 后端。

对于 APISIX，使用 Admin API 创建流路由：

curl "http://127.0.0.1:9180/apisix/admin/stream_routes/mqtt-route" -X PUT \
  -H "X-API-KEY: ${ADMIN_API_KEY}" \
  -H "Content-Type: application/json" \
  -d @- <<'EOF'
{
  "server_port": 9100,
  "plugins": {
    "mqtt-proxy": {
      "protocol_name": "MQTT",
      "protocol_level": 4
    }
  },
  "upstream": {
    "type": "roundrobin",
    "nodes": {
      "mqtt-broker-1.aic.svc.cluster.local:1883": 1
    }
  }
}
EOF

打开终端并直接订阅第一个 Broker：

docker exec mqtt-broker-1 \
  mosquitto_sub -h 127.0.0.1 -p 1883 \
  -V mqttv311 -t docs/mqtt

在另一个终端中，通过网关监听器发布消息：

docker exec mqtt-broker-1 \
  mosquitto_pub -h "$GATEWAY_CONTAINER" -p 9100 \
  -V mqttv311 -t docs/mqtt -m "Hello through the gateway"

订阅端应输出 Hello through the gateway。

代理 MQTT 5 流量​

本示例配置协议名称为 MQTT、协议级别为 5 的流路由，以接收 MQTT 5 CONNECT 数据包并通过网关代理消息。使用其他协议级别的连接将被拒绝。

对于 APISIX，使用 Admin API 创建流路由：

curl "http://127.0.0.1:9180/apisix/admin/stream_routes/mqtt-route" -X PUT \
  -H "X-API-KEY: ${ADMIN_API_KEY}" \
  -H "Content-Type: application/json" \
  -d @- <<'EOF'
{
  "server_port": 9100,
  "plugins": {
    "mqtt-proxy": {
      "protocol_name": "MQTT",
      "protocol_level": 5
    }
  },
  "upstream": {
    "type": "roundrobin",
    "nodes": {
      "mqtt-broker-1.aic.svc.cluster.local:1883": 1
    }
  }
}
EOF

使用 MQTT 5 客户端订阅：

docker exec mqtt-broker-1 \
  mosquitto_sub -h 127.0.0.1 -p 1883 \
  -V mqttv5 -t docs/mqtt-v5

通过网关发布 MQTT 5 消息：

docker exec mqtt-broker-1 \
  mosquitto_pub -h "$GATEWAY_CONTAINER" -p 9100 \
  -V mqttv5 -t docs/mqtt-v5 -m "Hello MQTT 5"

订阅端应输出 Hello MQTT 5。

按 MQTT 客户端 ID 进行负载均衡​

本示例使用 mqtt_client_id 配置一致性哈希上游，使同一客户端的连接继续到达同一个 Broker，并使用 MQTT 5 流量验证亲和性。

当客户端发送空 ID 时，网关会改用客户端 IP 进行一致性哈希。

对于 APISIX，使用 Admin API 创建流路由：

curl "http://127.0.0.1:9180/apisix/admin/stream_routes/mqtt-route" -X PUT \
  -H "X-API-KEY: ${ADMIN_API_KEY}" \
  -H "Content-Type: application/json" \
  -d @- <<'EOF'
{
  "server_port": 9100,
  "plugins": {
    "mqtt-proxy": {
      "protocol_name": "MQTT",
      "protocol_level": 5
    }
  },
  "upstream": {
    "type": "chash",
    "key": "mqtt_client_id",
    "nodes": {
      "mqtt-broker-1.aic.svc.cluster.local:1883": 1,
      "mqtt-broker-2.aic.svc.cluster.local:1883": 1
    }
  }
}
EOF

在不同终端中分别直接订阅两个 Broker：

docker exec mqtt-broker-1 \
  mosquitto_sub -h 127.0.0.1 -p 1883 \
  -V mqttv5 -t docs/balance

在第二个终端中订阅另一个 Broker：

docker exec mqtt-broker-2 \
  mosquitto_sub -h 127.0.0.1 -p 1883 \
  -V mqttv5 -t docs/balance

使用两个客户端 ID 通过网关发布消息：

docker exec mqtt-broker-1 \
  mosquitto_pub -h "$GATEWAY_CONTAINER" -p 9100 \
  -V mqttv5 -i publisher-1 \
  -t docs/balance -m "from publisher-1"

docker exec mqtt-broker-1 \
  mosquitto_pub -h "$GATEWAY_CONTAINER" -p 9100 \
  -V mqttv5 -i publisher-2 \
  -t docs/balance -m "from publisher-2"

消息可能到达相同或不同的 Broker。重复任一发布命令，验证在上游集合保持不变时，相同客户端 ID 会持续到达同一个 Broker。