上一篇我们讲了 如何创建一个基本的Newlife网络服务端 这边我们来讲一下如何解决粘包的问题
在上一篇总我们注册了Newlife的管道处理器 ，我们来看看他是如何实现粘包处理的
svr.Add<ReciveFilter>();//粘包处理管道
首先看一下我们设备的上传数据协议
设备上报的数据包头包含了固定的包头包尾，整个包的数据长度，设备编号。
包头：板卡类型，帧类型 2个字节 0x01 0x70
帧长度： 为两个字节 并且数据的字节序为 高字节在前 ，C#正常默认为低字节在前。
设备号：15位的ASCII 字符串
包尾： 两个字节 0x0D 0x0A 固定
下面来解决粘包的问题
Newlife网络库提供了几种常见的封包协议来解决粘包的问题，其中有一个 LengthFieldCodec解码器 这个解码器以长度字段作为头部 恰好符合我们的需求，我们就以这个解码器稍作改造来解决我们的粘包问题吧
由于这个解码器是适用于 只包含包头和包体的数据结构，且长度为包体的长度，而我们的协议 是包含包头包体包尾，并且帧长度为整个包的长度，长度为高位在前的数据结构，所以我们需要对整个解码器稍微做一些改造来符合我们的数据结构 。
我们来看下代码 其中
#region 属性
/// <summary>长度所在位置</summary>
public Int32 Offset { get; set; }=2;
/// <summary>长度占据字节数，1/2/4个字节，0表示压缩编码整数，默认2</summary>
public Int32 Size { get; set; } = 2;
/// <summary>过期时间，超过该时间后按废弃数据处理，默认500ms</summary>
public Int32 Expire { get; set; } = 500;
#endregion
在我们的协议中可以看到 设置了数据包的长度位置，长度占据的字节数,下面我们来获取一下整个包的长度
/// <summary>从数据流中获取整帧数据长度</summary>
/// <param name="pk"></param>
/// <param name="offset"></param>
/// <param name="size"></param>
/// <returns>数据帧长度（包含头部长度位）</returns>
protected Int32 GetLength(Packet pk, Int32 offset, Int32 size)
{
if (offset < 0) return pk.Total - pk.Offset;
// 数据不够，连长度都读取不了
if (offset >= pk.Total) return 0;
// 读取大小
var len = 0;
switch (size)
{
case 2:
var lenArry = pk.ReadBytes(offset, 2);
//高位在前，反转数组，获取长度
Array.Reverse(lenArry);
len = lenArry.ToUInt16();
break;
default:
throw new NotSupportedException();
}
// 判断后续数据是否足够
if (len > pk.Total) return 0;
return len;
}
获取长度后我们就可以从数据流中读取一个完整的包了
/// <summary>解码</summary>
/// <param name="context"></param>
/// <param name="pk"></param>
/// <returns></returns>
protected override IList<Packet> Decode(IHandlerContext context, Packet pk)
{
var ss = context.Owner as IExtend;
var mcp = ss["CodecItem"] as CodecItem;
if (mcp == null) ss["CodecItem"] = mcp = new CodecItem();
var pks = ParseNew(pk, mcp, 0, ms => GetLength(ms, Offset, Size), Expire);
// 跳过头部长度
var len = Offset + Math.Abs(Size);
foreach (var item in pks)
{
item.Set(item.Data, item.Offset + len, item.Count - len);
//item.SetSub(len, item.Count - len);
}
return pks;
}
#region 粘包处理
/// <summary>分析数据流，得到一帧数据</summary>
/// <param name="pk">待分析数据包</param>
/// <param name="codec">参数</param>
/// <param name="getLength">获取长度</param>
/// <param name="expire">缓存有效期</param>
/// <returns></returns>
protected IList<Packet> ParseNew(Packet pk, CodecItem codec, int startIndex, Func<Packet, Int32> getLength, Int32 expire = 5000)
{
var _ms = codec.Stream;
var nodata = _ms == null || _ms.Position < 0 || _ms.Position >= _ms.Length;
var list = new List<Packet>();
// 内部缓存没有数据，直接判断输入数据流是否刚好一帧数据，快速处理，绝大多数是这种场景
if (nodata)
{
if (pk == null) return list.ToArray();
var idx = 0;
while (idx < pk.Total)
{
//var pk2 = new Packet(pk.Data, pk.Offset + idx, pk.Total - idx);
var pk2 = pk.Slice(idx);
var len = getLength(pk2);
if (len <= 0 || len > pk2.Count) break;
pk2.Set(pk2.Data, startIndex, len);
//pk2.SetSub(0, len);
list.Add(pk2);
idx += len;
}
// 如果没有剩余，可以返回
if (idx == pk.Total) return list.ToArray();
// 剩下的
//pk = new Packet(pk.Data, pk.Offset + idx, pk.Total - idx);
pk = pk.Slice(idx);
}
if (_ms == null) codec.Stream = _ms = new MemoryStream();
// 加锁，避免多线程冲突
lock (_ms)
{
// 超过该时间后按废弃数据处理
var now = TimerX.Now;
if (_ms.Length > _ms.Position && codec.Last.AddMilliseconds(expire) < now)
{
_ms.SetLength(0);
_ms.Position = 0;
}
codec.Last = now;
// 合并数据到最后面
if (pk != null && pk.Total > 0)
{
var p = _ms.Position;
_ms.Position = _ms.Length;
pk.WriteTo(_ms);
_ms.Position = p;
}
// 尝试解包
while (_ms.Position < _ms.Length)
{
//var pk2 = new Packet(_ms.GetBuffer(), (Int32)_ms.Position, (Int32)_ms.Length);
var pk2 = new Packet(_ms);
var len = getLength(pk2);
// 资源不足一包
if (len <= 0 || len > pk2.Total) break;
// 解包成功
pk2.Set(pk2.Data, startIndex, len);
//pk2.SetSub(0, len);
list.Add(pk2);
_ms.Seek(len, SeekOrigin.Current);
}
// 如果读完了数据，需要重置缓冲区
if (_ms.Position >= _ms.Length)
{
_ms.SetLength(0);
_ms.Position = 0;
}
return list;
}
}
粘包处理管道完成后，就可以在Recive中去处理一个完整的数据包啦，我来解析一下这个状态的数据并且来保存设备连接
首先定义一个字典项用来保存设备的连接信息.设备号，连接的SessionId
/// <summary>
/// newLife连接保持
/// </summary>
private Dictionary<string, int> OnLineClients = new Dictionary<string, int>();
由于我们的数据中 帧类型不同的请求中帧类型是不一样的 所以解析数据需要做区分处理 我们来或者状态上传信息中的设备号并且和连接关联
private Dictionary<string, int> OnLineClients = new Dictionary<string, int>();
private object _lock=new object();
private void Recive(object sender, ReceivedEventArgs e)
{
INetSession session = (INetSession)sender;
var pk = e.Message as Packet;
if (pk.Count == 0)
{
XTrace.WriteLine("数据包解析错误");
return;
}
try
{
//数据包
var respBytes = pk.Data;
//获取帧类型
var dataTypeBytes = respBytes[1];
if (dataTypeBytes == 0x70)
{
//数值
byte[] deviceNoByte = new byte[15];
Buffer.BlockCopy(respBytes, 4, deviceNoByte, 0, 15); //从缓冲区里读取包头的字节
string deviceNo = Encoding.ASCII.GetString(deviceNoByte);
XTrace.WriteLine("设备编号：" + deviceNo); //保存连接信息 SaveClientConnection(deviceNo, session.ID);
//获取设备号后保存连接信息
} //支付宝
}
catch (Exception ex)
{
XTrace.WriteLine(ex.Message);
}
}
/// <summary>
/// 保存在线信息
/// </summary>
/// <param name="deviceNo"></param>
/// <param name="sessionId"></param>
private void SaveClientConnection(string deviceNo, int sessionId)
{
lock (_lock)
{
if (OnLineClients.ContainsKey(deviceNo))
{
OnLineClients[deviceNo] = sessionId;
}
else
{
OnLineClients.Add(deviceNo,sessionId);
}
}
}
好了数据粘包问题解决啦同时保存了设备连接信息，下面来解决如何定时检查测试在线状态。
原文地址：https://www.cnblogs.com/yushuo/p/10286472.html
时间： 2024-10-10 01:56:44