Rust MQTT Redis 持久存储

pa ho-mqtt-redis

一个库，用于将 Redis 作为 Paho MQTT Rust 客户端持久存储。

MQTT 是一种轻量级、分布式消息系统，特别适合网络连接不可靠的设备。它实现高质量服务的一部分方式是，客户端可以使用持久存储来保留正在传输的消息，直到远程系统确认收到为止。这样，即使客户端应用程序崩溃并重新启动，消息也可以被确认或重新发送。

通常，在使用持久性时，消息会保存到磁盘上。在某些较小的基于闪存的设备（如嵌入式 Linux 系统）上，这可能不是存储消息的最有效方式，因为连续写入可能会过早地磨损闪存芯片。

本地客户端设备上运行的 Redis 服务器可以作为消息的方便存储。Paho 库的用户定义持久机制使用键/值访问，这完美地映射到任何类似的 API，如 Redis。在这种上下文中使用它需要不到 100 行代码。

请注意，这仅适用于在 MQTT 客户端应用程序相同主机上运行的 本地 Redis 服务器。尝试在网络中持久化消息是没有意义的，因为这会在持久化级别引入网络问题和延迟。但本地 Redis 服务已在生产中得到证实。

pa ho MQTT Rust 库

pa ho MQTT Rust 库是 Paho C 库的包装器。它可以通过 crates.io 包含在项目中，包名为 paho-mqtt。将其添加到 Cargo.toml 文件的依赖项部分

[dependencies]
paho-mqtt = "0.12"
paho-mqtt-redis = "0.3"

源代码仓库位于 GitHub 上

https://github.com/eclipse/paho.mqtt.rust

Rust Redis 客户端

此库使用 "redis" crate 与 Redis 服务器通信。它在 Cargo.toml 文件中列为依赖项。项目主页可在

https://github.com/mitsuhiko/redis-rs

请注意，此客户端假设Redis在本地计算机上运行，绑定到localhost并使用默认的Redis端口。由于其主要目的是防止不可靠的网络连接，因此将其用作持久存储库可能没有太多意义。因此，使用本地服务似乎是正确的选择。但是，要更新RedisClient构造函数 RedisPersistence::new() 以指定另一个服务器或端口是非常简单的。

MQTT持久化模型

Paho Rust库包含一个特质，可用于提供用户定义的持久化

pub trait ClientPersistence {
    fn open(&mut self, client_id: &str, server_uri: &str) -> MqttResult<()>;
    fn close(&mut self) -> MqttResult<()>;

    fn put(&mut self, key: &str, buffers: Vec<&[u8]>) -> MqttResult<()>;
    fn get(&mut self, key: &str) -> MqttResult<Vec<u8>>;

    fn remove(&mut self, key: &str) -> MqttResult<()>;
    fn keys(&mut self) -> MqttResult<Vec<String>>;
    fn clear(&mut self) -> MqttResult<()>; 
    fn contains_key(&mut self, key: &str) -> bool;
}

这些操作与提供诸如哈希图之类的键/值存储所提供的操作密切相关。Redis可以使用哈希类型一对一地实现这些操作，操作包括 hset、hget、hdel、hkeys、del 和 hexists。发送给函数的 key 字符串可以作为Redis哈希字段。

哈希的名称（主键）通过将 client_id 和 server_uri 连接在一起生成，这些名称在调用 open() 时提供。

该库的大部分内容都致力于实现一个 RedisPersistence 结构体，该结构体用于实现与Redis服务器一起使用的 ClientPersistence 特质。

MQTT Rust客户端

使用Redis持久化相对简单。在 示例 文件夹中有一个示例应用程序 redis_persist_pub.rs，演示了其在示例中的使用。

执行以下操作

1. 创建一个 RedisPersistence 结构体实例。
2. 创建一个MQTT CreateOptions 结构体实例，指定RedisPersistence对象作为用户定义的持久化。
3. 创建MQTT客户端，使用这些选项。

可以这样做

let persistence = RedisPersistence::new();

let opts = mqtt::CreateOptionsBuilder::new()
    .server_uri("tcp://127.0.0.1:1883")
    .user_persistence(persistence)
    .finalize();

let cli = mqtt::AsyncClient::new(opts).unwrap_or_else(|e| {
    println!("Error creating the client: {:?}", e);
    process::exit(1);
});

然后库将自动使用Redis持久化根据需要保存和恢复消息和其他数据。