物联网海量设备心跳注册，脱网清除——多线程高并发互斥锁落地
物联网海量设备心跳注册，脱网清除——多线程高并发互斥锁落地
1.应用背景
在物联网应用场景中，需要维护很多个设备的连接，比如基于tcp socket通信的长连接，目的是为了获取设备采集的信息，反向控制设备的数字开关或者模拟量。我们把这些tcp长连接都放入了基于线程安全的concurrentdictionary激活字典表中，ip地址作为key，设备箱领域模型作为value。我们需要把激活设备箱的字典表维护好，需要将超时没有心跳的设备，我们可以称之为脱网设备，给清理出激活字典表，写入到脱网告警字典表中去。当脱网设备下次再有心跳时，可以再次移入到激活字典表中，从而再产生恢复告警，进行一系列其他动作。
2.整体框架
2.1.心跳注册框架
2.1.1.海量设备
因为要模拟海量设备的tcp场景，我们利用模拟器生成了12000台模拟设备。8台真实设备。
2.1.2.心跳上报handler流程
详细心跳上报流程详见上述框架图
- 第一次建立tcp长连接，并且上报心跳报文；
- socket缓存会先处理tcp中存在的粘包，具体方法可参见此篇博文 tcp粘包处理现象及其解决方案——基于newlife.net网络库的管道式帧长粘包处理方法
- 然后会触发onreceive中的e事件，从而传入粘包处理后的message；
- 判断包有效性，因为这方面比较简单，根据不同协议写一个类来处理即可，这里不再展开；
- 包有效载荷的crc判断，具体实现可参见此篇博文 基于modbus三种crc16校验方法的性能对比；
- 包类型解析（这里特指解析出心跳包）；
- 心跳包解析，具体可参见这两篇博文。深入浅出c#结构体——封装以太网心跳包的结构为例, 类与结构体性能对比测试——以封装网络心跳包为例
- 最终将设备新增激活字典表（第一次心跳）或者在激活字典表刷新心跳时间（非第一次心跳）。
突然发现我可以写一个物联网的采集系统的系列了，组织一个目录。希望自己坚持下去吧。
2.2.脱网清理框架
2.2.1.激活字典表清理脱网设备方法
原理很简单，遍历字典表中超过设置的检测周期，筛选到一个字典的ienumerable中去，然后在激活字典表中删除对应超时key（这里就是指ip地址）即可。当然这里的_internal周期可以*n，多个周期，自行在配置文件中设置即可，配置文件如下：
"ipboxnumstaticinternal": 12
public static void deletedeadboxfromactivebox(in _internal)
{
{
var outtime = datetime.now.addseconds(-_internal);
var iboxtimeoutlist = iboxactivedictionary.where(q => (outtime > q.value.updatetime));//.select(x=> iboxactivedictionary[x.key]) ;
foreach (var item in iboxtimeoutlist)
{
iboxactivedictionary.remove(item.key);
}
}
}
2.2.2.脱网清理流程图
这里主要开启了一个系统定时器，主动会去调用清理脱网设备方法，调用时间间隔即ipboxnumstaticinternal。代码如下：
public void systemtimerstart()
{
var interval = readtheinternalfromsetting();
_systemtimer = new timer(state =>
{
iboxactivedicmanager.deletedeadboxfromactivebo(_internal);
console.writeline("{1},激活设备数量：{0}\n",iboxactivedicmanager.iboxactivedictionary.count,datetime.now);
}, null, interval, interval);
console.writeline("pemscom采集系统时钟已经开启");
loggerhelper.info("pemscom采集系统时钟已经开启");
}
/// <summary>
/// 配置文件读入时间间隔方法
/// </summary>
/// <returns></returns>
private int readtheinternalfromsetting()
{
_internal = int.parse(appsettings.app(new string[] {"ipboxnumstaticinternal" }));
console.writeline("pemscom采集系统时钟配置参数已经读");
loggerhelper.info("pemscom采集系统时钟配置参数已经读");
return convert.toint32(timespan.fromsecond(_internal).totalmilliseconds);
}
3.多线程与高并发说明
3.1.多线程说明
这里会有很多的线程让cpu来轮片执行，比如：
- 12008个receive事件触发线程；
- 定时清除脱网设备线程；
- 主线程，监控命令行输入，并执行对应的命令；
举个实际的例子，以图为证
12008台设备，每秒处理接受网络包的峰峰值是9218个包，就是在某一秒，cpu共轮片执行了9218个线程。比如是双核4线程的，则9218/4=2304.5。即cpu在1秒轮片执行了2305次。即0.43毫秒就轮片执行一次。
3.2.高并发说明
其实3.1已经解释了高并发。在某一秒，需要处理的接收事件有接近1万件。而这一时刻的执行顺序是无序的，9218里的这么多线程，我们不知道哪个先执行，哪个后执行。如果不认为地加一些逻辑控制，比如我们今天要介绍的互斥锁，就会出现一些异常现象。
4.多线程高并发造成的异常现象
这里只描述现象，原因会在下面5.分析异常原因 做具体描述。
4.1.空引用
异常所在的位置：心跳处理类如下。
public class hearthandler
{
static string _deviceindex = appsettings.app(new string[] { "deviceindex" });
private static iboxactive iboxactive;
public static void register(tcpheartpacket heartpacket,int sessid)
{
uint32 ip;
uint64 mac;
if (_deviceindex == "ip")
{
ip =(uint32)bitconverter.touint32(heartpacket.ip, 0);
if (iboxactivedicmanager.getboxactive(ip, out iboxactive) != true)
{
iboxactivedicmanager.iboxactivedictionary.tryadd(ip, iboxactive);
iboxactive.sessid = sessid;
}
}
else
{
mac = (uint64)bitconverter.touint64(heartpacket.mac, 0);
if (iboxactivedicmanager.getboxactive(mac, out iboxactive) != true)
{
iboxactivedicmanager.iboxactivedictionary.tryadd(mac, iboxactive);
iboxactive.sessid = sessid;
}
}
//引用类型，智能指针，使用方便
iboxactive.updatetime = datetime.now;
}
}
4.2.字典表里元素赋值不成功
/// <summary>
/// 查询激活设备箱字典中是否有存在上报的设备箱,
/// 存在返回true，不存在返回false，并且新建好设备箱模型
/// </summary>
/// <param name="mac"></param>
/// <param name="iboxactive"></param>
/// <returns></returns>
public static bool getboxactive(uint32 ip, out iboxactive iboxactive)
{
if (iboxactivedictionary.trygetvalue(ip, outiboxactive))
{
return true;
}
iboxactive = new iboxactive();
iboxactive.ip = ip;
if (iboxactive.ip != ip)
{
loggerhelper.error(string.format("实例化赋值不成功.iboxactive.ip:{0};ip{1}", iboxactive.ip, ip));
}
return false;
}
有没有感觉很奇怪，上一句都赋值了，下一句对比就不相等。但是在多线程大并发里就是有这种可能，下面会详细分析。
4.3.统计设备总数不正确
因为12008台大并发时很容易出错，所以改成了1000台。如下统计数据会有出错情况，这同样也是因为多线程高并发引起的错误。
5.分析异常原因
5.1.造成空引用的原因
其实第4的三点原因都是同一个原因造成，所以在5.1会详细阐述，5.2，,5.3只做简单阐述。这里敲下黑板，分析多线程高并发的异常问题，程序运行的特点就是见缝就插，就像个老司机一样，概括起来就是线程与线程之间的无序性。比如我们设备心跳线程正在更新设备心跳时间的时候。脱网清理线程就把该设备给清理掉了。如此一来，时间没法赋值给空对象（已被脱网线程给清理）。因此只能报空引用异常，对没错，就是这么简单，耗费了我很长时间去debug跟思考这个异常。
5.2.设备ip赋值不成功原因
同样，在创建了设备实例之后，ip赋值完成，刚好脱网清除设备线程运行清除了设备，当对比的时候，引用原来的地址，字典的原来地址已经存了其他设备箱的ip，所以ip地址不相等。
5.3.统计设备总数不正确原因
原因其实是5.2造成的，没法成功注册，当然数量就不对啦。
6.解决思路
就是当我在创建激活设备实例（第一次心跳注册）或者更新心跳时间的时候（非第一次注册），不要让无序的脱网清除线程运行。敲黑板:就是保证心跳处理注册过程的原子性。对，其实这里很像关系型数据库的事务，原子性。原子性就是对抗程序无序造成异常的有力武器。我们可以在注册心跳处理方法上加个互斥锁，让编译器跟运行时去安排更加合理的执行顺序。
7.代码实现
代码很简单。
//定义一把锁
public static mutex activeipboxdicmutex = new mutex();
//设备箱注册加锁。异常全部消除
iboxactivedicmanager.activeipboxdicmwaitone();
hearthandler.register(tcpheartpacsessionid);
iboxactivedicmanager.activeipboxdicmreleasemutex();
这里插入一下事务的使用，也是很类似的，把我们的主业务加中中间，类比方便大家理解记忆。就像夹心饼干（瞎扯）。
unitofwork.begintransaction();
// adds new device
unitofwork.devicerepository.add(device);
// commit transaction
unitofwork.commit();
当然也可以给设备箱脱网清除线程加锁。
iboxactivedicmanager.activeipboxdicmutex.waitone();
iboxactivedicmanager.deletedeadboxfromactivebox(_internal);
iboxactivedicmanager.activeipboxdicmutex.releasemutex();
考虑到脱网清除线程会损耗部分性能，我也测试了去掉该锁的情况，也不会有第4的3个异常，至此问题全部解决。
8.小结
-
模拟设备数量小测不出这个问题，如此看出海量设备的重要性，因为现实情况肯定会出现以上三个问题，而且都是很严重很致命的问题。好的测试方法可以把问题扼杀在摇篮中；
-
多线程高并发时容易出现这样那样的异常，要怀着敬畏之心去思考，去解决问题；
版权声明：本文为博主原创文章，遵循 cc 4.0 by-sa 版权协议，转载请附上原文出处链接和本声明。