Apache NMS C#中MQTT文件持久化实现方法咨询

我之前在项目里刚好研究过Apache NMS MQTT的文件持久化，确实不像Eclipse Paho那样一眼就能找到配置，但其实它提供了对应的实现，只是需要稍微调整一下配置和代码。下面是具体的实现步骤和注意事项：

1. 先确保依赖正确 #

首先要通过NuGet安装官方的Apache.NMS.MQTT包，这是基础：

Install-Package Apache.NMS.MQTT

2. 核心配置：指定文件持久化存储 #

NMS MQTT提供了MqttFilePersistence类，专门用于文件级的持久化存储。你只需要指定一个本地目录，它会自动在这个目录下生成持久化所需的文件。

完整代码示例 #

using Apache.NMS;
using Apache.NMS.MQTT;
using Apache.NMS.MQTT.Persistence;

namespace NmsMqttPersistenceDemo
{
    class Program
    {
        static void Main(string[] args)
        {
            // 1. 配置本地文件持久化目录（确保有读写权限）
            var persistenceDir = @"C:\mqtt-local-persistence";
            var filePersistence = new MqttFilePersistence(persistenceDir);

            // 2. 创建连接工厂并绑定持久化实现
            var factory = new MqttConnectionFactory("tcp://your-mqtt-broker:1883")
            {
                Persistence = filePersistence,
                // 必须设置唯一的ClientId！持久化状态和ClientId绑定
                ClientId = "nms-persistent-client-001"
            };

            // 3. 建立连接并创建持久化会话
            using (var connection = factory.CreateConnection())
            {
                connection.Start();

                // 使用ClientAcknowledge模式，确保消息确认后才会从本地存储移除
                using (var session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge))
                {
                    // 4. 创建生产者，设置持久化投递模式
                    var destination = session.GetDestination("test/persistent/topic");
                    var producer = session.CreateProducer(destination);
                    // 关键：设置投递模式为Persistent，确保消息被持久化
                    producer.DeliveryMode = DeliveryMode.Persistent;

                    // 发送持久化消息
                    var message = session.CreateTextMessage("这是一条会被文件持久化的MQTT消息");
                    producer.Send(message);
                    Console.WriteLine("持久化消息已发送");

                    // 5. 创建消费者，订阅主题并处理消息
                    var consumer = session.CreateConsumer(destination);
                    consumer.Listener += (receivedMsg) =>
                    {
                        var textMsg = receivedMsg as ITextMessage;
                        Console.WriteLine($"收到持久化消息：{textMsg.Text}");
                        // 确认消息，NMS会自动从本地存储中删除该记录
                        receivedMsg.Acknowledge();
                    };

                    Console.WriteLine("等待接收消息...按回车退出");
                    Console.ReadLine();
                }
            }
        }
    }
}

3. 关键注意事项 #

• 必须设置唯一ClientId：持久化的状态（比如未确认的消息）是和ClientId绑定的，如果客户端重启时使用相同的ClientId，NMS会自动恢复之前的持久化状态。
• 目录权限：确保指定的持久化目录有读写权限，否则会抛出IO异常。
• 消息确认：只有当你调用msg.Acknowledge()后，NMS才会从本地文件存储中删除对应的消息记录；如果客户端意外崩溃，重启后会重新接收未确认的消息。
• 自定义持久化：如果默认的MqttFilePersistence不能满足需求（比如需要加密存储），你可以实现IMqttPersistence接口，自定义自己的文件持久化逻辑。

内容的提问来源于stack exchange，提问作者Krishna Prasad