项目地址：https://github.com/hnlyf168/DotNet.Framework
昨天晚上大致测试了下 ，490个客户端（一个收一个发） 平均估计每个每秒60个包 使用mqtt协议 发送一个guid的字符串 服务器转发每秒大约1.2-1.3w
cpu 占用：25% 一下
内存好像都在50m以下
1、协议简介
MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。
具体就不在这里记录了，写这个服务端和客户端也只是为了更加深入的学习mqtt协议。
2、mqtt的几种控制报文类型
|
名字 |
值 |
报文流动方向 |
描述 |
|
Reserved |
0 |
禁止 |
保留 |
|
CONNECT |
1 |
客户端到服务端 |
客户端请求连接服务端 |
|
CONNACK |
2 |
服务端到客户端 |
连接报文确认 |
|
PUBLISH |
3 |
两个方向都允许 |
发布消息 |
|
PUBACK |
4 |
两个方向都允许 |
QoS 1消息发布收到确认 |
|
PUBREC |
5 |
两个方向都允许 |
发布收到（保证交付第一步） |
|
PUBREL |
6 |
两个方向都允许 |
发布释放（保证交付第二步） |
|
PUBCOMP |
7 |
两个方向都允许 |
QoS 2消息发布完成（保证交互第三步） |
|
SUBSCRIBE |
8 |
客户端到服务端 |
客户端订阅请求 |
|
SUBACK |
9 |
服务端到客户端 |
订阅请求报文确认 |
|
UNSUBSCRIBE |
10 |
客户端到服务端 |
客户端取消订阅请求 |
|
UNSUBACK |
11 |
服务端到客户端 |
取消订阅报文确认 |
|
PINGREQ |
12 |
客户端到服务端 |
心跳请求 |
|
PINGRESP |
13 |
服务端到客户端 |
心跳响应 |
|
DISCONNECT |
14 |
客户端到服务端 |
客户端断开连接 |
|
Reserved |
15 |
禁止 |
保留 |
2.1、协议头
每个MQTT控制报文都包含一个固定报头，
固定报头的格式
|
Bit |
7 |
6 |
5 |
4 |
3 |
2 |
1 |
0 |
|
byte 1 |
MQTT控制报文的类型 |
用于指定控制报文类型的标志位 |
||||||
|
byte 2... |
剩余长度 |
|||||||
剩余长度的计算方式：
剩余长度（Remaining Length）表示当前报文剩余部分的字节数，包括可变报头和负载的数据。剩余长度不包括用于编码剩余长度字段本身的字节数。
剩余长度字段使用一个变长度编码方案，对小于128的值它使用单字节编码。更大的值按下面的方式处理。低7位有效位用于编码数据，最高有效位用于指示是否有更多的字节。因此每个字节可以编码128个数值和一个延续位（continuation bit）。剩余长度字段最大4个字节
例如，十进制数64会被编码为一个字节，数值是64，十六进制表示为0x40,。十进制数字
321(=65+2*128)被编码为两个字节，最低有效位在前。第一个字节是 65+128=193。注意最高位为1表示后面至少还有一个字节。第二个字节是2。
.net 计算方式代码如下：
/// <summary>
/// 获取一个长度数据
/// </summary>
/// <returns></returns>
protected virtual Result<int> ReadLength()
{
var result = this.Socket.ReceiveBytes();
if (!result.Success)
{
WriteLog("获取mqtt 长度失败");
return new Result<int>(result);
}
var msgType = result.Data[];
var msgLength = msgType & ;//取低7为的值，因为可变长度有效值只有低7位，第8位用来标识下一个字节是否属于长度字节
var leftBit = ;
while (msgType >> == )//判断最高位是否为1，如果为1则说明后面的1个字节也是属于长度字节
{
result = this.Socket.ReceiveBytes();
if (!result.Success)
{
WriteLog("获取mqtt 长度失败");
return new Result<int>(result);
}
msgType = result.Data[];
msgLength = ((msgType & ) << leftBit) | msgLength;// 因为mqtt 可变长度的字节是低位在前，所以新取到的长度要左移取到的次数*7位在|原来的长度。
leftBit += ;
}
return msgLength;
}
2.2、CONNECT – 连接服务端
协议格式
可看到所需要的参数，于是定义一个连接信息类来保存
/// <summary>
/// mqtt 连接信息。
/// </summary>
public class MQTTConnectInfo
{
/// <summary>
/// 客户端编号
/// </summary>
public virtual string ClientId { get; set; }
/// <summary>
/// 用户名
/// </summary>
public virtual string UserName { get; set; }
/// <summary>
/// 用户密码
/// </summary>
public virtual string Password { get; set; }
/// <summary>
/// 遗嘱保留
/// </summary>
public virtual bool WillRetain { get; set; }
/// <summary>
/// 遗嘱QoS
/// </summary>
public virtual Qos WillQos { get; set; }
/// <summary>
/// 遗嘱标志
/// </summary>
public virtual bool WillFlag { get; set; }
/// <summary>
/// 是否清除对话。
/// </summary>
public virtual bool CleanSession { get; set; }
/// <summary>
/// 保持连接
/// <para>警告：这里的单位是秒</para>
/// </summary>
public virtual ushort KeepAlive { get; set; } = ;
}
然后就是代码按协议格式组装好
代码如下：
/// <summary>
/// 获取包完整的字节
/// </summary>
/// <returns></returns>
public override byte[] ToBytes()
{
var list = new List<byte>();
var mqttBytes = ProtocolName.ToBytes(Encoding.ASCII);//协议名称：固定位MQTT
list.Add((byte)(mqttBytes.Length >> ));
list.Add((byte)(mqttBytes.Length & ));
list.AddRange(mqttBytes); list.Add(Version);//协议版本
list.Add(ConnectFlag);//连接标识 list.Add((byte)(KeepAlive >> ));//心跳值
list.Add((byte)(KeepAlive & )); var clientIdBytes = ClientId.ToBytes(Encoding.ASCII);//客户端编号
list.Add((byte)(clientIdBytes.Length >> ));
list.Add((byte)(clientIdBytes.Length & ));
list.AddRange(clientIdBytes); if (HasUserName)//是否包含用户名
{
var userNameBytes = UserName.ToBytes(Encoding.ASCII);
list.Add((byte)(userNameBytes.Length >> ));
list.Add((byte)(userNameBytes.Length & ));
list.AddRange(userNameBytes);
}
if (HasPassword)//是否包含用户密码
{
var passwordBytes = Password.ToBytes(Encoding.ASCII);
list.Add((byte)(passwordBytes.Length >> ));
list.Add((byte)(passwordBytes.Length & ));
list.AddRange(passwordBytes);
}
Data = list.ToArray();
list.Clear();
return base.ToBytes();
}
连接回复包格式：
|
描述 |
7 |
6 |
5 |
4 |
3 |
2 |
1 |
0 |
|||||||||
|
连接确认标志 |
Reserved 保留位 |
1 SP |
|||||||||||||||
|
byte 1 |
0 |
0 |
0 |
0 |
0 |
0 |
0 |
X |
|||||||||
|
连接返回码 |
|||||||||||||||||
|
byte 2 |
X |
X |
X |
X |
X |
X |
X |
X |
|||||||||
|
描述 |
7 |
6 |
5 |
4 |
3 |
2 |
1 |
0 |
|||||||||
|
连接确认标志 |
Reserved 保留位 |
1 SP |
|||||||||||||||
|
byte 1 |
0 |
0 |
0 |
0 |
0 |
0 |
0 |
X |
|||||||||
|
连接返回码 |
|||||||||||||||||
|
byte 2 |
X |
X |
X |
X |
X |
X |
X |
X |
|||||||||
，犹豫代码比较简单这里就不贴了
2.3、PUBLISH – 发布消息
PUBLISH控制报文是指从客户端向服务端或者服务端向客户端传输一个应用消息
|
主题长度 16位 2字节 |
|
主题内容N |
|
如果QoS大于0 则有一个消息Id 16位 2字节 |
|
剩余的N 是消息的主题 |
|
描述 |
7 |
6 |
5 |
4 |
3 |
2 |
1 |
0 |
|
|
Topic Name 主题名 |
|||||||||
|
byte 1 |
Length MSB (0) |
0 |
0 |
0 |
0 |
0 |
0 |
0 |
0 |
|
byte 2 |
Length LSB (3) |
0 |
0 |
0 |
0 |
0 |
0 |
1 |
1 |
|
byte 3 |
‘a’ (0x61) |
0 |
1 |
1 |
0 |
0 |
0 |
0 |
1 |
|
byte 4 |
‘/’ (0x2F) |
0 |
0 |
1 |
0 |
1 |
1 |
1 |
1 |
|
byte 5 |
‘b’ (0x62) |
0 |
1 |
1 |
0 |
0 |
0 |
1 |
0 |
|
报文标识符 |
|||||||||
|
byte 6 |
报文标识符 MSB (0) |
0 |
0 |
0 |
0 |
0 |
0 |
0 |
0 |
|
byte 7 |
报文标识符 LSB (10) |
0 |
0 |
0 |
0 |
1 |
0 |
1 |
0 |
代码按协议格式组装如下：
/// <summary>
/// 开始组装包。
/// </summary>
protected override void Packaging()
{
var topicBytes = Topic.ToBytes();//主题数据
Data = new byte[topicBytes.Length + BodyBytes.Length + (QoS > ? : )];
Data[] = (byte)(topicBytes.Length >> );
Data[] = (byte)(topicBytes.Length & );
topicBytes.CopyTo(Data, );
if (QoS > )
{
Data[topicBytes.Length + ] = (byte)(Identifier >> );
Data[topicBytes.Length + ] = (byte)(Identifier & );
}
BodyBytes.CopyTo(Data, Data.Length - BodyBytes.Length);//复制消息内容
topicBytes = null;
}
后面的格式我就不在一一写出来了 ，附上一个mqtt协议文档
这里还有一个非常重要的自己写的一个Socket 辅助类，主要实现高性能的读取和发送消息，已整包的形式，这样避免粘包等问题
/// <summary>
/// Berkeley 套接字 辅助
/// </summary>
public abstract class SocketClient<Package>
where Package : IDataPackage
{
private Socket m_Socket;
private Timer timerHeartbeat;
private System.Net.EndPoint remoteEndPoint;
/// <summary>
/// 客户端唯一标识
/// </summary>
public virtual long Id { get; set; }
/// <summary>
/// Berkeley 套接字。
/// </summary>
public virtual Socket Socket { get => m_Socket; protected set { m_Socket = value; remoteEndPoint = m_Socket.RemoteEndPoint; } } /// <summary>
/// 客户端的远程信息。
/// </summary>
public virtual System.Net.EndPoint RemoteEndPoint { get => remoteEndPoint; }
/// <summary>
/// 心跳线程
/// </summary>
protected virtual Timer TimerHeartbeat { get => timerHeartbeat; } /// <summary>
/// 心跳时间。
/// </summary>
public virtual int KeepAlive { get; set; } = ;
/// <summary>
/// 初始化
/// </summary>
/// <param name="socket"></param>
protected SocketClient(Socket socket)
{ Socket = socket;
}
/// <summary>
/// 初始化
/// </summary>
protected SocketClient()
{ }
/// <summary>
/// 读取一个完整的包。
/// </summary>
/// <returns></returns>
protected abstract DotNet.Result<Package> ReceivePackage();
/// <summary>
/// 开始循环读取消息。
/// </summary>
public virtual void OnReceive()
{
while (!IsClose)
{
try
{
OnHeartbeatTimer();
var bytesResult = ReceivePackage();
if (bytesResult.Success)
{
OnNewDataPackage(bytesResult);
}
else
{
WriteLog($"接收包时错误，错误内容：{bytesResult.Message}");
if (bytesResult.Code == -)
{
this.Close();
}
}
}
catch (Exception ex)
{
WriteErrorLog($"接收包时异常", ex);
}
}
Close();
}
/// <summary>
/// 当接收到
/// </summary>
/// <param name="bytesResult"></param>
protected virtual void OnNewDataPackage(Result<Package> bytesResult)
{
try
{
// 这里使用异步会有一个问题，就是如果一个客户端while(true)在发消息，会导致服务器线程被一个客户端占满而无法处理其他的客户端。
OnHandleDataPackage(bytesResult.Data);
}
catch (Exception ex)
{
WriteErrorLog($"客户端处理包时报错", ex);
}
} #if NET40
/// <summary>
/// 启用异步读取
/// </summary>
/// <returns></returns>
public virtual Task OnReceiveAsync()
{
return Task.Factory.StartNew(OnReceive);
}
#else
/// <summary>
/// 启用异步读取
/// </summary>
/// <returns></returns>
public virtual async Task OnReceiveAsync()
{
await Task.Run(() =>
{
OnReceive();
});
}
#endif private bool m_IsClose;
/// <summary>
/// 是否已经关闭
/// </summary>
public virtual bool IsClose => m_IsClose;
/// <summary>
/// 关闭连接，并退出当前线程
/// </summary>
public virtual void Close(int timeout = )
{
lock (this)
{
if (!IsClose)
{
m_IsClose = true;
WriteLog($"关闭连接"); OnClose();
//真正关闭，避免二次关闭
}
}
Socket?.Close(timeout);
Socket?.Dispose();
timerHeartbeat?.Dispose();
}
/// <summary>
/// 关闭连接并退出。
/// </summary>
protected abstract void OnClose();
/// <summary>
/// 设置心跳计数器
/// </summary>
protected virtual void OnHeartbeatTimer()
{
if (timerHeartbeat == null)
{
timerHeartbeat = new Timer(OnHeartbeatTimerCallback, this, KeepAlive, KeepAlive);
}
else
{
timerHeartbeat.Change(KeepAlive, KeepAlive);
}
}
/// <summary>
/// 心跳实际到达后触发，改方法又心跳计数器执行。
/// </summary>
/// <param name="state"></param>
protected virtual void OnHeartbeatTimerCallback(object state)
{
WriteLog($"客户端{KeepAlive}s未发包，已丢弃");
Close();
}
/// <summary>
/// 写入日志。
/// </summary>
/// <param name="text">日志内容</param>
public virtual void WriteLog(string text)
{
// Log.WriteLog($" 连接{RemoteEndPoint}-{text}");
}
/// <summary>
/// 写入错误信息到日志。
/// </summary>
/// <param name="text">错误信息描述</param>
/// <param name="exception">异常信息</param>
public virtual void WriteErrorLog(string text, Exception exception = null)
{
// Log.WriteErrorLog($" 连接{RemoteEndPoint}-{text}", exception);
}
/// <summary>
/// 写入日志。
/// </summary>
/// <param name="text">日志内容</param>
/// <param name="args"></param>
public virtual void WriteLog(string text, params object[] args)
{
WriteLog(string.Format(text, args));
}
/// <summary>
/// 开始处理接收的包
/// </summary>
/// <param name="dataPackage"></param>
protected abstract void OnHandleDataPackage(Package dataPackage);
/// <summary>
/// 发送数据
/// </summary>
/// <param name="bytes"></param>
public virtual Result SendBytes(byte[] bytes)
{
lock (this)
{
if (!IsClose)
{
try
{
Socket.Send(bytes);
return true;
}
catch (Exception ex)
{
WriteErrorLog($"发送数据{bytes.ToBase64String()}", ex);
if (!Socket.Connected)
{
Close();
}
} }
}
return false;
}
}
其mqtt的Socket实现子类如下：
/// <summary>
/// mqtt 服务器连接过来的客户端。
/// </summary>
public class MQTTSocketClient : DotNet.Net.SocketClient<MQTTDataPackage>
{
/// <summary>
/// 表示mqtt服务器。
/// </summary>
public virtual MQTTServer TcpServer { get; set; }
/// <summary>
/// 获取一个值，该值指示客户端是否发送过了连接协议包。
/// </summary>
public virtual bool IsConnect { get; protected set; }
private readonly List<TopicDataPackage> subscribeTopics = new List<TopicDataPackage>();
/// <summary>
/// 订阅主题。
/// </summary>
public TopicDataPackage[] SubscribeTopics { get => subscribeTopics.ToArray(); }
/// <summary>
/// 当前消息序号
/// </summary>
public virtual ushort Identifier { get; set; }
/// <summary>
/// 客户端连接编号
/// </summary>
public virtual string ClientId { get; set; }
/// <summary>
/// 客户端唯一连接id
/// </summary>
public override long Id
{
get
{
if (long.TryParse(ClientId, out long id))
{
base.Id = id;
}
else
{
base.Id = ClientId.GetHashCode();
}
return base.Id;
}
set
{
ClientId = value.ToString();
}
}
/// <summary>
/// 写日志。
/// </summary>
/// <param name="text"></param>
public override void WriteLog(string text)
{
if (ClientId != null)
{
text = $"客户端编号：{ClientId}：{text}";
}
// base.WriteLog(text);
}
/// <summary>
/// 使用<see cref="Socket"/>客户端初始化。
/// </summary>
/// <param name="socket"></param>
public MQTTSocketClient(Socket socket) : base(socket)
{ }
/// <summary>
/// 关闭服务端连接
/// </summary>
protected override void OnClose()
{
Console.WriteLine($"{ClientId}关闭连接");
} /// <summary>
/// 处理收到的包
/// </summary>
/// <param name="dataPackage"></param>
protected override void OnHandleDataPackage(MQTTDataPackage dataPackage)
{ try
{ WriteLog($"收到{dataPackage.MessageType} 包， QoS level:{dataPackage.QoS}"); if (IsConnect && dataPackage.MessageType != MessageType.Connect)
{
WriteLog($"收到{dataPackage.MessageType} 包， QoS level:{dataPackage.QoS} ,但连接尚未登录，被抛弃");
this.Close(); } switch (dataPackage.MessageType)
{
case MessageType.Connect:
OnConnect(dataPackage);
break;
case MessageType.Subscribe:
OnSubscribe(dataPackage);
break;
case MessageType.PingRequest:
OnPingRequest(dataPackage);
break;
case MessageType.Publish:
OnPublishPackage(dataPackage);
break;
case MessageType.UnSubscribe:
OnUnSubscribe(dataPackage);
break;
case MessageType.Disconnect:
this.Close();
break;
}
}
catch (Exception ex)
{ }
dataPackage = null; }
#if NET40
/// <summary>
/// 当收到发布消息
/// </summary>
/// <param name="dataPackage"></param>
protected virtual Task OnPublishPackage(MQTTDataPackage dataPackage)
{
return Task.Factory.StartNew(() =>
{
#else
/// <summary>
/// 当收到发布消息
/// </summary>
/// <param name="dataPackage"></param>
/// <returns></returns>
protected virtual async Task OnPublishPackage(MQTTDataPackage dataPackage)
{
await Task.Run(() =>
{
#endif
try
{
PublishDataPackage publishDataPackage = new PublishDataPackage(dataPackage);
var result = OnPublish(publishDataPackage);
if (dataPackage.QoS > )
{
var package = new MQTTDataPackage() { MessageType = MessageType.PublishAck, Data = new byte[] { (byte)(publishDataPackage.Identifier >> ), (byte)(publishDataPackage.Identifier & ), } };
if (dataPackage.QoS == )
{
if (!result.Success)
{
package.Data[] = ;
}
//SendPackage(package);
}
}
}
catch (Exception ex)
{ }
});
}
/// <summary>
/// 当客户发布消息。
/// </summary>
/// <param name="message"></param>
/// <returns></returns>
protected virtual Result OnPublish(PublishDataPackage message)
{
WriteLog($"客户端{message.ClientId}发布消息{message.Topic},QoS{message.QoS}。内容：{message.Text}");
try
{
foreach (var client in TcpServer.Clients)
{
foreach (var topic in client.SubscribeTopics)
{
if (MqttTopicFilterComparer.IsMatch(message.Topic, topic.Topic))
{
var temp = message.Clone();
temp.QoS = ;// Math.Min(message.QoS, topic.QoS);//mqtt协议规定，取订阅主题和发送主题中最小的qos值。
client.Publish(temp);
}
}
}
}
catch (Exception ex)
{ }
return true;
}
/// <summary>
/// 发布消息。
/// </summary>
/// <param name="message">要发布的消息。</param>
/// <returns></returns>
public virtual Result Publish(PublishDataPackage message)
{
message.Identifier = ++Identifier;
this.SendPackage(message);//目前不校验，qos 直接发送
return true;
}
/// <summary>
/// 当客户端发送了ping 请求
/// </summary>
/// <param name="dataPackage"></param>
protected virtual void OnPingRequest(MQTTDataPackage dataPackage)
{
var package = new MQTTDataPackage() { MessageType = MessageType.PingResponse };
SendPackage(package);
}
/// <summary>
/// 发生订阅消息
/// </summary>
/// <param name="dataPackage"></param>
private void OnSubscribe(MQTTDataPackage dataPackage)
{ TopicDataPackage topicDataPackage = new TopicDataPackage(dataPackage);
var result = OnSubscribe(topicDataPackage);
var package = new SubscribeAckDataPackage() { Identifier = topicDataPackage.Identifier, Success = result.Success };
if (result.Success)
{
if (!subscribeTopics.Contains(topicDataPackage))
{
subscribeTopics.Add(topicDataPackage);
}
package.ValidQos = Qos.QoS2;//
}
SendPackage(package);
}
/// <summary>
/// 取消订阅消息
/// </summary>
/// <param name="dataPackage"></param>
private void OnUnSubscribe(MQTTDataPackage dataPackage)
{
TopicDataPackage topicDataPackage = new TopicDataPackage(dataPackage);
var result = OnUnSubscribe(topicDataPackage);
if (result.Success)
{
if (subscribeTopics.Contains(topicDataPackage))
{
subscribeTopics.Remove(topicDataPackage);
}
var package = new IdentifierAckDataPackage(MessageType.UnSubscribeAck) { Identifier = topicDataPackage.Identifier };
SendPackage(package);
}
}
/// <summary>
/// 当收到 取消订阅主题消息时。
/// </summary>
/// <param name="message"></param>
/// <returns></returns>
protected virtual Result OnUnSubscribe(TopicDataPackage message)
{
WriteLog($"客户端{message.ClientId} 取消订阅{message.Topic},QoS{message.QoS}");
return true;
}
/// <summary>
/// 当收到订阅主题消息时。
/// </summary>
/// <param name="message"></param>
/// <returns></returns>
protected virtual Result OnSubscribe(TopicDataPackage message)
{
WriteLog($"客户端{message.ClientId}订阅{message.Topic},QoS{message.RequestedQoS}");
return true;
}
/// <summary>
/// 当客户端发送连接请求时。
/// </summary>
/// <param name="dataPackage">连接请求的包</param>
private void OnConnect(MQTTDataPackage dataPackage)
{
ConnectDataPackage connectDataPackage = new ConnectDataPackage(dataPackage);
var result = OnClientConnect(connectDataPackage);
var client = TcpServer.GetClientById(connectDataPackage.ClientId);
if (client.Success)
{
client.Data.WriteLog($"新的客户端连接{this.RemoteEndPoint}上线，旧连接关闭");
client.Data.Close();
}
ClientId = connectDataPackage.ClientId;
this.KeepAlive = Convert.ToInt32(connectDataPackage.KeepAlive * * 1.5);
var package = new ConnectAckDataPackage() { Result = result };
SendPackage(package); }
/// <summary>
/// 发送一个标准的mqtt包到客户端连接。
/// </summary>
/// <param name="package"></param>
public virtual void SendPackage(MQTTDataPackage package)
{
WriteLog($"发送{package.MessageType}包,QOS：{package.QoS}");
this.SendBytes(package.ToBytes());
}
/// <summary>
/// 当客户端连接到服务验证是否可以连接
/// </summary>
/// <param name="message"></param>
/// <returns></returns>
protected virtual Result OnClientConnect(ConnectDataPackage message)
{
WriteLog($"客户端{message.ProtocolName}连接,客户端编号{message.ClientId},用户名：{message.UserName}，密码：{message.Password},CeanSession:{message.CeanSession}");
return true;
}
/// <summary>
/// 接收一个完整的包。
/// </summary>
/// <returns></returns>
protected override Result<MQTTDataPackage> ReceivePackage()
{ Result<byte[]> result;
Result<MQTTDataPackage> resultPackage = new Result<MQTTDataPackage>() { Success = false };
MQTTDataPackage package = new MQTTDataPackage() { ClientId = ClientId, RemoteEndPoint = RemoteEndPoint };
result = this.Socket.ReceiveBytes();
if (!result.Success)
{
WriteLog("获取mqtt 头 首字节失败");
this.Close();
return resultPackage;
}
package.Header = result.Data[];
var msgLengthResult = ReadLength();
if (!msgLengthResult.Success)
{
WriteLog(msgLengthResult.Message);
return resultPackage;
}
result = this.Socket.ReceiveBytes(msgLengthResult.Data);
if (!result.Success)
{
WriteLog($"获取数据长度{msgLengthResult.Data}内容失败");
return resultPackage;
}
package.Data = result.Data;
resultPackage.Data = package;
resultPackage.Success = true;
resultPackage.Message = "获取包成功";
return resultPackage;
}
/// <summary>
/// 获取一个长度数据
/// </summary>
/// <returns></returns>
protected virtual Result<int> ReadLength()
{
var result = this.Socket.ReceiveBytes();
if (!result.Success)
{
WriteLog("获取mqtt 长度失败");
return new Result<int>(result);
}
var msgType = result.Data[];
var msgLength = msgType & ;//取低7为的值，因为可变长度有效值只有低7位，第8位用来标识下一个字节是否属于长度字节
var leftBit = ;
while (msgType >> == )//判断最高位是否为1，如果为1则说明后面的1个字节也是属于长度字节
{
result = this.Socket.ReceiveBytes();
if (!result.Success)
{
WriteLog("获取mqtt 长度失败");
return new Result<int>(result);
}
msgType = result.Data[];
msgLength = ((msgType & ) << leftBit) | msgLength;// 因为mqtt 可变长度的字节是低位在前，所以新取到的长度要左移取到的次数*7位在|原来的长度。
leftBit += ;
}
return msgLength;
}
}