使用MASA全家桶从零开始搭建IoT平台（四）处理设备上行数据-获取并通知业务系统

前言

分析

方案

完全基于MQTT的方案
我们与设备约定，设备连接MQTT成功后，消息全部通过名称规则为"topic/{设备名称}"的Topic传递
那么我们可以订阅所有规则为"topic/{设备名称}"的Topic来获取该设备的上行数据。
EMQX中还支持通配符，我们可以通过订阅"topic/+",来获取所有设备的上行数据。关于通配符的更多内容，请参考官方文档

https://www.emqx.io/docs/zh/v5/deploy/cluster/introduction.html#主题树-主题匹配通配符

这个方案可行，但是存在单点故障，如果我们的IoT Hub出现了故障，或者数据流量很大导致Hub出现瓶颈，导致最新上报的数据无法及时有效地被处理，可能会影响业务
我们想到可以分配多个Hub，我们可以在设备发布数据时随机生成1-10个Topic分组，“topic/1/+”…“topic/10/+” ，这样我们可以启动多个Hub，然后每个Hub分别订阅一个主题，这样压力会平分到10个Hub server中，这样做虽然可以减少系统瓶颈的可能，但是如果某一个hub挂了，还是会有一部分设备上报的数据受影响，直到该hub恢复。
如果可以弹性的伸缩Hub订阅，支持负载均衡就好了。

共享订阅

EMQX 支持共享订阅，多个订阅者订阅同一主题，EMQX Broker会按照一定的分发策略将消息发给订阅者，在这个层面上实现订阅者的负载均衡。
共享订阅只需要在订阅时加上前缀即可（发布的时候无需添加）
在原有主题的基础上，添加 $share 前缀即可为一组订阅端启用共享订阅，而且也是支持通配符 “#” 和 “+”。

带群组的共享订阅

以 $share/<group-name> 为前缀的共享订阅是带群组的共享订阅：
group-name 可以为任意字符串，属于同一个群组内部的订阅者将以负载均衡接收消息，但 EMQX 会向不同群组广播消息。
例如，假设订阅者 s1，s2，s3 属于群组 g1，订阅者 s4，s5 属于群组 g2。
那么当 EMQX 向这个主题发布消息 msg1 的时候：
EMQX 会向两个群组 g1 和 g2 同时发送 msg1
s1，s2，s3 中只有一个会收到 msg1
s4，s5 中只有一个会收到 msg1

均衡分发策略

EMQX 支持很多不同的平衡策略（MQTT协议中并没有明确的规范平衡策略）
平衡策略可以在全局或每组中指定。可以在 etc/emqx.conf 文件中修改

1. 1.全局策略可以在 broker.shared_subscription_strategy 配置中设置。

   2.配置 broker.shared_subscription_group.$group_name.strategy 为每组策略。

使用共享订阅的方式，可以根据数据量动态的添加和减少Hub数量，不存在单点故障，扩展性好。

数据格式

上行数据分两部分，Payload和Metadata
Metadata部分，我们在MQTTX中我们可以通过点击Meta来添加

• 1、Payload指消息所携带的数据本身，多数情况为传感器数据，可以是任何格式，例如二进制或者JSON字符串，格式是业务和嵌入式之间约定的，Hub部分不解析Payload只是将它传递到业务系统
• 2、Metadata相当于我们http请求中的head数据，可以添加发布者的身份信息或者消息的唯一ID等等。据本人了解，貌似不是所有的MQTT平台都支持设备端自定义Metadata,或者说很多通讯模组已经配置好了Metadata，不支持自定义，所以这里我们暂时不在Metadata中配置任何信息。

实施流程

1. 我们在MASA.IoT.Hub项目中创建一个服务HostedService
2. 以mqtt client的形式连接EMQX的mqtt，并订阅所有设备的Topic
3. 将消息原封不动的封装，并发送给业务系统（MASA.IoT.Core项目中的api接口）

编写代码

using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Formatter;

namespace MASA.IoT.Common.Helper
{
    public class MqttHelper
    {
        private MqttFactory _mqttFactory;
        private IMqttClient _mqttClient;
        private MqttClientOptions _mqttClientOptions;
        private MqttClientSubscribeOptions _mqttClientSubscribeOptions;

        public MqttHelper(string mqttUrl, string clientID, string userName, string passWord)
        {
            _mqttFactory = new MqttFactory();
            _mqttClient = _mqttFactory.CreateMqttClient();
            _mqttClientOptions = new MqttClientOptionsBuilder()
                                  .WithTcpServer(mqttUrl)
              .WithCredentials(userName, passWord).WithProtocolVersion(MqttProtocolVersion.V500).Build();

            _mqttClientOptions.ClientId = clientID;
        }

        /// <summary>
        /// 连接并订阅Topic
        /// </summary>
        /// <param name="callback"></param>
        /// <param name="topic"></param>
        /// <returns></returns>
        public async Task ConnectClient(Func<MqttApplicationMessageReceivedEventArgs, Task> callback, string topic) 
        {
            _mqttClientSubscribeOptions = _mqttFactory.CreateSubscribeOptionsBuilder()
                .WithTopicFilter(f => { f.WithTopic(topic); })
                .Build();

            var response = await _mqttClient.ConnectAsync(_mqttClientOptions, CancellationToken.None);
            if (response.ResultCode == MqttClientConnectResultCode.Success)
            {
                Console.WriteLine($"The MQTT client with topic:{topic} is connected.");
                await Task.Delay(500);
                _mqttClient.ApplicationMessageReceivedAsync += callback;
                await _mqttClient.SubscribeAsync(_mqttClientSubscribeOptions, CancellationToken.None);
            }
        }

        /// <summary>
        /// 断开连接
        /// </summary>
        /// <returns></returns>
        public async Task Disconnect_Client() 
        {
            if (_mqttClient.IsConnected)
            {
                await _mqttClient.DisconnectAsync();

                Console.WriteLine("The MQTT client is Disconnected.");
            }
        }
    }
}

using Dapr.Client;
using MASA.IoT.Common;
using MASA.IoT.Common.Helper;
using Microsoft.Extensions.Options;
using MQTTnet.Client;

namespace MASA.IoT.Hub;

public class MQHostedService : IHostedService
{
    private readonly HubAppSettings _appSettings;
    private readonly DaprClient _daprClient;
    public MQHostedService(IOptions<HubAppSettings> appSettings)
    {
        _daprClient = new DaprClientBuilder().Build();
        _appSettings = appSettings.Value;
    }

    /// <summary>
    /// 开始                
    /// </summary>
    /// <param name="cancellationToken"></param>
    /// <returns></returns>
    public async Task StartAsync(CancellationToken cancellationToken)
    {
        var mqttHelper = new MqttHelper(_appSettings.MqttSetting.MqttUrl, "IoTHub", _appSettings.MqttSetting.UserName, _appSettings.MqttSetting.Password);
 
        await mqttHelper.ConnectClient(CallbackAsync, _appSettings.MqttSetting.Topic);

    }
    private async Task CallbackAsync(MqttApplicationMessageReceivedEventArgs e)
    {
        var deviceDataPointStr = System.Text.Encoding.Default.GetString(e.ApplicationMessage.PayloadSegment);
        
        Console.WriteLine(deviceDataPointStr);
        var pubSubOptions = new PubSubOptions
        {
            DeviceName = e.ApplicationMessage.Topic[6..],
            Msg = deviceDataPointStr,
            PubTime = new DateTimeOffset(DateTime.Now).ToUnixTimeMilliseconds(),
            TrackId = Guid.NewGuid()
        };                            
        try
        {
            await _daprClient.PublishEventAsync("pubsub", "DeviceMessage", pubSubOptions);
        }
        catch (Exception ex)
        {
            Console.WriteLine(ex.Message);
        }

    }
    /// <summary>
    /// 结束
    /// </summary>
    /// <param name="cancellationToken"></param>
    /// <returns></returns>
    public Task StopAsync(CancellationToken cancellationToken)
    {
        return Task.CompletedTask;
    }
}

appsettings.json内容
{
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "Microsoft.AspNetCore": "Warning"
    }
  },
  "AllowedHosts": "*",
  "MqttSetting": {
    "MqttUrl": "192.120.5.204",
    "UserName": "IoTHub",
    "Password": "123456",
    "Topic": "$share/IotHub/topic/+"
  },
  "DaprOptions": {
    "AppId": "masa-iot-service-hub",
    "AppPort": 18001,
    "AppIdSuffix": "",
    "DaprGrpcPort": 20241,
    "DaprHttpPort": 20242
  }
}

builder.Services.AddHostedService<MQHostedService>();

这部分不了解的可以参考第一章的Dapr搭建教程

using Dapr;
using MASA.IoT.Common;
using Microsoft.AspNetCore.Mvc;

namespace MASA.IoT.Core.Controllers
{

    [Route("api/[controller]")]
    [ApiController]
    public class DeviceMqttController : ControllerBase
    {
        [Topic("pubsub", "DeviceMessage")]
        [HttpPost("DeviceMessage")]
        public void DeviceMessageAsync([FromBody] PubSubOptions pubSubOptions)
        {
            Console.WriteLine($"Subscriber received, DeviceName:{pubSubOptions.DeviceName},Msg:{pubSubOptions.Msg}");
        }
    }
}

https://www.emqx.com/zh/blog/introduction-to-mqtt-qos

完整代码在这里：https://github.com/sunday866/MASA.IoT-Training-Demos