数据桥接

桥接是一种连接多个 MQTT 消息中间件的策略，其特性与集群模式显著不同。在桥接模式下，节点之间不进行主题树或路由表的复制操作。桥接模式的核心职能包括：

根据预定的规则，将消息转发至指定的桥接节点；
对桥接节点上的特定主题进行订阅，并在接收到消息后在本地节点或集群内进行传递和转发。

MQTT over TCP 桥接

本节将介绍 MQTT over TCP 数据桥接相关的配置参数。

配置示例

hcl

bridges.mqtt.emqx1 = {
  server = "mqtt-tcp://127.0.0.1:1883"    # MQTT 服务器地址
  proto_ver = 4                           # MQTT 协议版本
  clientid = "bridge_client"              # 桥接的客户端 ID
  keepalive = "60s"                       # 桥接的保活间隔时间（s）
  clean_start = false                     # 清除会话
  username = "username"                   # 桥接用户名
  password = "passwd"                     # 桥接密码
  will = {                                # 遗嘱消息相关配置
  	topic = "will_topic"                  # Will 主题
  	qos = 1                               # Will QoS
  	retain = false                        # 是否应保留遗嘱消息
  	payload = "will_message"              # Will payload
  	properties = {                        # Will 消息属性
    	payload_format_indicator = 0
    	message_expiry_interval = 0
    	content_type = ""
    	response_topic = ""
    	correlation_data = ""
    	will_delay_interval = 0
    	user_property = {
      	key1 = "value1"
      	key2 = "value2"
    	}
  	}
  }
  ssl = {                                 # SSL 配置
    key_password = "yourpass"             # SSL 密钥密码
    keyfile = "/etc/certs/key.pem"        # SSL 密钥文件
    certfile = "/etc/certs/cert.pem"      # SSL 证书文件
    cacertfile = "/etc/certs/cacert.pem"  # SSL CA 证书文件
    verify_peer = true                    # 启用 SSL 对端证书验证

    sni = "broker.emqx.io"                # SSL Server Name Indication
  }
  
  forwards = [                            # 要转发到远端 MQTT 服务器的主题
    {
      remote_topic = "fwd/topic1"
      local_topic = "topic1"
      suffix = "/emqx"
    },
    {
      remote_topic = "fwd/topic2"
      local_topic = "topic2"
      prefix = "emqx/"
    }
  ]     
  subscription = [                        # 要从远端 MQTT 服务器订阅的主题
    {
      remote_topic = "cmd/topic1"
      local_topic = "topic3"
      qos = 1
      retain = 2                          # 重载标志位
    },
    {
      remote_topic = "cmd/topic2"
      local_topic = "topic4"
      qos = 2
    }
  ]
  max_parallel_processes = 2              # 最大并行进程数
  max_send_queue_len = 32                 # 消息发送队列的最大长度
  max_recv_queue_len = 128                # 消息接收队列的最大长度
}

通过以上配置，EMQX Edge 可以建立到远端 MQTT 服务器的 MQTT over TCP 桥接，并使用 Will Message 和 SSL 安全通信。

配置项

bridges.mqtt.<name>：桥接名称。
server：桥接目标 Broker 的地址 URL。示例：
- MQTT over TCP 桥接：mqtt-tcp://127.0.0.1:1883
- 经 SSL 加密的 MQTT over TCP 桥接：tls+mqtt-tcp://127.0.0.1:8883
- MQTT over QUIC 桥接：mqtt-quic://54.75.171.11:14567
proto_ver：指定 MQTT 协议版本：可选值：
- 5：MQTT v5
- 4：MQTT v3.1.1
- 3： MQTT v3.1
clientid：桥接客户端 ID，默认 NULL 为自动生成随机 ID。
keepalive：保活间隔时间。
clean_start：清除会话。注意：有些 IoT 平台会要求该项设为 false。
username：登录用户名。
password：登录密码。
ssl：SSL/TLS 相关配置项：
- key_password：TLS 私钥密码。
- keyfile：TLS 私钥数据。
- certfile：TLS Cert 证书数据。
- cacertfile：TLS CA 证书数据。
- verify_peer：指定是否验证远端 MQTT 服务器证书。
- sni：指定 TLS 握手期间发送的 Server Name Indication（SNI）值，通常设置为远端 Broker 的主机名。
forwards：转发到远端 MQTT 服务器的主题数组，包括：
- remote_topic：发布到远端 Broker 的主题。将 remote_topic 留空可保留原始主题。
- local_topic：需要转发到远端 MQTT 服务器的本地主题。
- qos：覆盖原始 PUBLISH 消息的 QoS 等级。该项可选。
- suffix：添加到远端主题的后缀字符串。如果 remote_topic 为空，则添加到原始主题。
- prefix：添加到远端主题的前缀字符串。如果 remote_topic 为空，则添加到原始主题。
subscription：需要从远程 MQTT 服务器订阅的主题对象数组：
- remote_topic：用于订阅远端 Broker 的主题过滤器。
- local_topic：用于主题映射。将 local_topic 留空可保留远端 Broker 消息中的原始主题。
- qos：SUBSCRIBE 报文中的 QoS。该项必填。
- retain：重载标志位。
- retain_as_published：MQTT v5 可选特性，Retain As Published。
- retain_handling：MQTT v5 可选特性，Retain Handling。
- suffix：添加到本地主题的后缀字符串。如果 local_topic 为空，则添加到原始主题。
- prefix：添加到本地主题的前缀字符串。如果 local_topic 为空，则添加到原始主题。
max_parallel_processes：接客户端并发数。
max_send_queue_len：最大发送队列长度。
max_recv_queue_len：最大接收队列长度。

MQTT 5

如果选择使用 MQTT 5.0 协议（proto_ver = 5），EMQX Edge 还支持以下配置项：

连接相关：

配置项	说明	取值范围
`conn_property.maximum_packet_size`	最大报文长度	1 - 4294967295
`conn_properties.receive_maximum`	QoS 1 和 QoS 2 消息的最大接收数量，仅在当前连接下有效。如未配置，将使用默认值 65535。	1 - 65535
`conn_properties.topic_alias_maximum`	主题别名最大长度	0 - 65535
`conn_properties.request_problem_information`	请求问题信息： - 如设为 0，服务器仅可在 PUBLISH、CONNACK 或 DISCONNECT 包中插入问题信息。如违反该规则，客户端则将断开连接并返回协议错误信息。 - 如设为 1，则不限制包的类型。	0 或 1
`conn_properties.request_response_information`	请求响应信息： - 如设为 0，服务器禁止返回响应信息。 - 如设为 1，服务器可以在 CONNACK 包中返回响应信息。	0 或 1
`conn_properties.session_expiry_interval`	会话过期间隔： - 如设为 0，会话将在网络连接关闭后结束。 - 如设为 4294967295 (UINT_MAX)，会话则永不过期。	0 - 4294967295
`conn_properties.user_property`	用户属性键值对。	Map[key(String) - value(String)]

订阅相关

配置项	说明	取值范围
`sub_properties.identifier`	订阅标识符	1 ~ 268,435,455
`sub_properties.user_property`	用户属性	Map[key(String) - value(String)]*

遗嘱消息（Will Message）

遗嘱消息是 MQTT 为那些可能出现意外断线的设备提供的将遗嘱优雅地发送给第三方的能力，意外断线包括但不限于：

因网络故障或网络波动，设备在保持连接周期内未能通讯，连接被服务端关闭
设备意外掉电
设备尝试进行不被允许的操作而被服务端关闭连接，例如订阅自身权限以外的主题等

MQTT over QUIC 桥接

本节将介绍 MQTT over QUIC 数据桥接相关的配置参数。QUIC 最初由 Google 开发，后来被互联网工程任务组（IETF）采纳为全球标准。它是一种新的传输协议，提供更快的连接建立速度。通过 MQTT over QUIC 数据桥接，我们可以充分发挥 QUIC 协议在 IoT 场景中的优势。

配置示例

hcl

bridges.mqtt.emqx1 = {
  server = "mqtt-quic://127.0.0.1:14567"  # MQTT 服务器地址
  proto_ver = 4                           # MQTT 协议版本
  clientid = "bridge_client"              # 桥接的客户端 ID
  keepalive = "60s"                       # 桥接的保活间隔时间（s）
  clean_start = false                     # 清除会话
  username = "username"                   # 桥接用户名
  password = "passwd"                     # 桥接密码
  quic_keepalive = "120s"                 # 使用 QUIC 桥接的 ping 间隔
  quic_idle_timeout = "120s"              # 使用 QUIC 桥接的连接最大过期时间
  quic_discon_timeout = "20s"             # QUIC 桥接等待连接 ACK 最大时间
  quic_handshake_timeout = "60s"          # QUIC 握手最大超时时间
  quic_send_idle_timeout = "2s"           # QUIC 传输层重置拥塞控制算法的等待超时时间
  quic_initial_rtt_ms = "800ms"           # 初始 RTT 估计时间
  quic_max_ack_delay_ms = "100ms"         # 发送 ACK 之前接收数据后等待时长
  hybrid_bridging = false                 # 混合桥接模式开关
  quic_multi_stream = false               # 多流传输开关
  quic_qos_priority = true                # 高优先级发送 QOS 1 或 2 的消息
  quic_0rtt = true                        # 0RTT 开关，用于快速重新建立连接
  forwards = [                            # 要转发到远端 MQTT 服务器的主题
    {
      remote_topic = "fwd/topic1"
      local_topic = "topic1"
      qos = 1
    },
    {
      remote_topic = "fwd/topic2"
      local_topic = "topic2"
      qos = 2
    }
  ]     
  subscription = [                        # 要从远端 MQTT 服务器订阅的主题
    {
      remote_topic = "cmd/topic1"
      local_topic = "topic3"
      qos = 1
    },
    {
      remote_topic = "cmd/topic2"
      local_topic = "topic4"
      qos = 2
    }
  ]
  max_parallel_processes = 2              # 最大并行进程数
  max_send_queue_len = 32                 # 消息发送队列的最大长度
  max_recv_queue_len = 128                # 消息接收队列的最大长度
}

配置项

本部分重点介绍 MQTT over QUIC 桥接相关的配置项，其他配置项可参考 MQTT over TCP 桥接。

Server：桥接的 MQTT 服务器地址，例如 mqtt-quic://54.75.171.11:14567
quic_keepalive：QUIC 传输层保活时间，缺省为 120 秒。
quic_idle_timeout：QUIC 连接最大过期时间，超时后，连接将被断开。0 表示永不超时，缺省为 120 秒。
quic_discon_timeout：QUIC 等待连接 ACK 最大时间，缺省为 20s。
quic_handshake_timeout：QUIC 握手最大超时时间，缺省为 60 秒。
quic_send_idle_timeout：传输层重置拥塞控制算法的等待超时时间，缺省为 60 秒。
quic_initial_rtt_ms：初始 RTT 估计时间，单位：ms，缺省为 800 ms。
quic_max_ack_delay_ms：发送 ACK 之前接收数据后等待时长，缺省为 100 ms。
hybrid_bridging：确认是否开启混合桥接模式；缺省为 false。
hybrid_servers：确认混合桥接URLs；缺省为 []。
quic_multi_stream：确认是否开始多流传输，本功能目前正在进一步验证中，不建议开启。缺省为 false。
quic_qos_priority：确认高优先级发送 QOS 1 或 2 的消息，QoS 0 消息的优先级不变。缺省为 true。
quic_0rtt：确认是否开启 0RTT QUIC 协议特性，用于快速重新建立连接。缺省为 true。

TIP

MQTT over QUIC 桥接暂不支持 SSL 相关配置。

MQTT 桥接缓存

EMQX Edge 允许在 nanomq.conf 文件中配置多个 MQTT 数据桥接。每个桥接实例都可以通过名称唯一标识。除各桥接自身配置外，EMQX Edge 还提供共享的 cache 组件，可配置一次并被多个桥接实例使用。

该缓存功能尤其适用于桥接负载较高，或远端 Broker 暂时不可用时临时保存 QoS 消息。

配置示例

hcl

## 第一个桥接客户端
bridges.mqtt.emqx1 {
  retry_qos_0 = true
  ......
  resend_interval = 5000    # Resend interval (ms)
  resend_wait = 3000
  cancel_timeout  = 10000
}

## 第二个桥接客户端
bridges.mqtt.emqx2 {
  retry_qos_0 = false
  ......
  resend_interval = 5000    # Resend interval (ms)
  resend_wait = 3000
  cancel_timeout  = 10000
}

bridges.mqtt.cache {
    disk_cache_size = 102400   # Max message limitation for caching
    mounted_file_path="/tmp/"  # Mounted file path
    flush_mem_threshold = 100  # The threshold of flushing messages to flash
}

配置项

retry_qos_0：指定是否缓存 QoS 0 消息。设置为 false 可为 QoS 1 和 QoS 2 消息保留磁盘空间，适用于需要优先级处理的场景。
disk_cache_size：指定 MQTT 桥接中可以缓存的消息的最大数量。0 表示不生效。
mounted_file_path：指定 MQTT 桥接缓存文件的挂载路径。
flush_mem_threshold：指定刷新消息到缓存文件的阈值。当消息数量达到阈值时，就会被刷新到缓存文件中。
resend_interval：指定消息重发间隔，单位为毫秒。该配置仅对桥接生效。它是每个桥接连接的定时器，也负责发送 PINGREQ、重发 SQLite 中缓存的消息和健康检查，请谨慎设置。
- 默认值：5000 ms。
resend_wait：指定消息发布后等待多久再重发，单位为毫秒。仅对桥接生效。
- 默认值：3000 ms。
cancel_timeout：指定取消 QoS ACK 动作前的最长等待时间，单位为毫秒。仅对桥接生效。动作取消后，该消息不会继续重试。
- 默认值：8000 ms。

QoS 消息处理说明

取消 QoS 消息并不意味着消息丢失，只表示客户端已停止等待远端 Broker 的 ACK。

在 0.22.4 之前的版本中没有该机制。如果桥接遇到繁忙的 I/O 线程（aio），可能会看到如下消息： bridging to xxxxx aio busy! msg lost! Ctx: xx 如果远端 Broker 一直没有响应，该消息会持续卡住。

为避免这种情况，引入了 cancel_timeout 作为安全窗口，避免无限期阻塞。它定义 QoS 消息在从重试队列丢弃前等待 ACK 的时间。

保证 QoS 消息重试（调优示例）

对于桥接中的 QoS 消息重试保证，可参考以下配置：

hcl

## 第一个桥接客户端
bridges.mqtt.emqx1 {
  ......
  keepalive = 30s           # Taking 30s keepalive as context
  max_send_queue_len = 512  # Give the in-flight window enough space for caching messages
  max_recv_queue_len = 512  # Give the in-flight window enough space for caching messages
  resend_interval = 5000    # Retry QoS messages every 5 seconds if no other action is blocking.
                            # Set this to at least 1/2 or 1/4 of the keepalive interval.
  resend_wait = 3000        # Wait before resending messages after they are published.
                            # Set this longer than keepalive to avoid duplicate QoS messages.
  cancel_timeout  = 10000   # Maximum time before canceling the ACK action.
                            # This is also the time window reserved for each QoS message.
                            # (cancel_timeout - resend_wait) / resend_wait > 1: retry at least once.
}

调优建议

QoS 消息会先进入飞行窗口（max_send_queue_len）。如果网络拥塞，它们会静默等待，直到有可用空间。
只有当飞行窗口已满且新的 QoS 消息持续到达时，消息才会被丢弃。
每条 QoS 消息的实际超时时间为：cancel_timeout - resend_wait。将该值除以 resend_interval，可估算可能的重试次数。
例如：(10000 - 3000) / 5000 ≈ 1.4 表示至少重试 1 次。

达到超时时间后，消息会从 QoS 等待队列中移除（不是从飞行窗口中移除），并且不会再继续重发。

EMQX Edge 使用 SQLite 提供缓存功能。有关配置详情，请参见 Cache。

客户端认证

授权

MQTT over TCP

MQTT over QUIC

安全

数据桥接

MQTT over TCP 桥接

配置示例

配置项

MQTT 5

遗嘱消息（Will Message）

MQTT over QUIC 桥接

配置示例

配置项

MQTT 桥接缓存

配置示例

配置项

保证 QoS 消息重试（调优示例）

调优建议

数据桥接 ​

MQTT over TCP 桥接 ​

配置示例 ​

配置项 ​

MQTT 5 ​

遗嘱消息 （Will Message） ​

MQTT over QUIC 桥接 ​

配置示例 ​

配置项 ​

MQTT 桥接缓存 ​

配置示例 ​

配置项 ​

保证 QoS 消息重试（调优示例） ​

调优建议 ​

数据桥接

MQTT over TCP 桥接

配置示例

配置项

MQTT 5

遗嘱消息（Will Message）

MQTT over QUIC 桥接

配置示例

配置项

MQTT 桥接缓存

配置示例

配置项

保证 QoS 消息重试（调优示例）

调优建议