# 保留消息

# 简介

MQTT 服务端收到 Retain 标志为 1 的 PUBLISH 报文时,会将该报文视为保留消息,除了被正常转发以外, 保留消息会被存储在服务端,每个主题下只能存在一份保留消息,因此如果已经存在相同主题的保留消息,则该保留消息被替换。

当客户端建立订阅时,如果服务端存在主题匹配的保留消息,则这些保留消息将被立即发送给该客户端。 借助保留消息,新的订阅者能够立即获取最近的状态,而不需要等待无法预期的时间,这在很多场景下是非常重要的。

EMQX 默认开启保留消息的能力和服务,可以在 etc/emqx.conf 中修改 mqtt.retain_availablefalse 来关闭保留消息的能力, 这样客户端将被禁止发送 Retain 标志为 1 的 PUBLISH 报文,否则,客户端将会收到原因码为 0x9A(不支持保留消息)的 DISCONNECT 报文。

保留消息的服务会存储和管理客户端发送的保留消息,并发送给相应的订阅者。

# 快速开始

# 设置

打开 Dashboard,选择 功能配置 中的 MQTT 项,然后选择 保留消息 页面中的 设置 栏即可

image

# 配置

配置项类型可取值默认值说明
存储方式enumram, discramram:仅储存在内存中;
disc:储存在内存和硬盘中。
最大存储条数integer>= 00保留消息的最大数量,0 表示没有限制。保留消息数量超出最大值限制后,可以替换已存在的保留消息,但不能为新的主题储存保留消息。
最大存储大小bytesize1MB保留消息的最大 Payload 值。Payload 大小超出最大值后 EMQ X 消息服务器会把收到的保留消息作为普通消息处理。
有效期duration保留消息的过期时间,0 表示永不过期。如果 PUBLISH 报文中设置了消息过期间隔,那么以 PUBLISH 报文中的消息过期间隔为准。
定时清理duration清理回收过期消息的间隔时间。

# 流控

在保留消息中可以设置消息的读取和派发速率。

当客户端订阅了包含通配符的主题时,它可能会匹配到大量的保留消息,如果没有速率控制,所有匹配到的消息将被一次性发送给订阅者,这可能会导致订阅者的进程因为内存大量激增而被强制关闭。

为了降低风险,可以使用 retainer.flow_control 设置保留消息的读取和派发速率, 例如:

# 每个订阅保留消息的会话每次会加载 10 条消息并投递 10 条消息,所有这些会话的总投递速率被限制为 100/s,而保留消息中的每个工作进程的派发速率被限制为 20/s (大多数情况下,并不需要配置 client 这级)
retainer {
  enable = true
  flow_control {
    batch_read_number = 10
    batch_read_deliver = 10
    batch_read_limiter {
      rate = "100/s"
      capacity = 100
      client {
        rate = "20/s"
        capacity = 20
      }
    }
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

配置说明:

字段类型默认值说明
batch_read_numberint0每次读取的消息数量(0 表示所有的)
batch_deliver_numberint0每次派发的消息数量(0 表示所有的)
batch_deliver_limiterlimiterundefined消息派发的速率限制器

关于速率限制器的详细设置,请参考分层速率控制系统 中的 监听器级连接级