實(shí)驗(yàn)三 物聯(lián)網(wǎng)中間件中實(shí)時數(shù)據(jù)預(yù)警及數(shù)據(jù)清洗

實(shí)驗(yàn)三物聯(lián)網(wǎng)中間件的創(chuàng)建及數(shù)據(jù)清洗3.1實(shí)驗(yàn)背景近年來，隨著物聯(lián)網(wǎng)概念提出以及RFID射頻識別技術(shù)在沃爾瑪供應(yīng)鏈管理上的成功運(yùn)用，使得以往僅用于軍事領(lǐng)域的RFID技術(shù)，憑借其成本低、體積小、無須接觸等特性越來越多的運(yùn)用在生產(chǎn)自動化、門禁、公路收費(fèi)、貨物跟蹤等民用領(lǐng)域和礦山的人員定位系統(tǒng)中。然而，由于受到各種環(huán)境和射頻技術(shù)本身的一些特點(diǎn)，RFID閱讀器識別標(biāo)簽時會發(fā)生各種錯誤識別現(xiàn)象：積極讀、消極讀、標(biāo)簽冗余、閱讀器冗余，這一些因素都會制約著RFID技術(shù)應(yīng)用進(jìn)一步發(fā)展。在傳感器中同樣會產(chǎn)生很多臟數(shù)據(jù)，例如錯讀，漏讀現(xiàn)象。對上述幾種臟數(shù)據(jù)，分別采用相應(yīng)的算法加以清洗。由于原始數(shù)據(jù)流具有無限性，因此在清洗算法中使用了滑動窗口技術(shù)。其中，采用平均值策略消除數(shù)據(jù)冗余：即將一分鐘的數(shù)據(jù)進(jìn)行平均，將平均值來取代這一分鐘內(nèi)產(chǎn)生的多個數(shù)據(jù)，這樣使得存入數(shù)據(jù)庫的數(shù)據(jù)量大大減少，而當(dāng)用戶做歷史查詢時，也能體現(xiàn)出數(shù)據(jù)變化的大致趨勢。移動平均值是對傳感器數(shù)據(jù)錯讀現(xiàn)象的平滑處理，將該數(shù)據(jù)和該數(shù)據(jù)的前后幾次數(shù)據(jù)求平均值，在很多情況下這個平均值就能更好的體現(xiàn)出這個時間點(diǎn)在時間段中的所要表達(dá)的數(shù)據(jù)?；瑒哟翱诓呗詣t可以消除積極讀的現(xiàn)象，消除積極讀實(shí)質(zhì)上是去除原始數(shù)據(jù)流中的噪音，算法通過統(tǒng)計(jì)滑動窗口中的標(biāo)簽個數(shù)來判斷標(biāo)簽是否為噪音。實(shí)時數(shù)據(jù)預(yù)警是判斷采集的數(shù)據(jù)值是否在規(guī)定的范圍內(nèi)，如果不在，就設(shè)置預(yù)警標(biāo)識，對實(shí)時數(shù)據(jù)進(jìn)行預(yù)警。3.2實(shí)驗(yàn)?zāi)康?、了解中間件數(shù)據(jù)清洗的意義；2、了解中間件的開發(fā)；3、掌握中間件對數(shù)據(jù)處理的流程與控制;4、掌握中間件數(shù)據(jù)清洗的幾種基本策略5、掌握中間件實(shí)時數(shù)據(jù)預(yù)警的方法3.3實(shí)驗(yàn)壞境VisualStudio20103.4實(shí)驗(yàn)內(nèi)容一、具體內(nèi)容1、中間件的開發(fā)2、編寫中間件數(shù)據(jù)清洗策略3、為中間件添加實(shí)時數(shù)據(jù)預(yù)警功能二、內(nèi)容詳解1、中間件包括以下5個功能（1）初始化工作，包括初始化一級緩存和8小時緩存；（2）數(shù)據(jù)采集工作，采集由OPCserver產(chǎn)生的數(shù)據(jù)（3）數(shù)據(jù)處理工作，采用適當(dāng)?shù)那逑床呗詫?shù)據(jù)進(jìn)行清洗處理。（4）數(shù)據(jù)轉(zhuǎn)存工作，將處理過的數(shù)據(jù)存入到數(shù)據(jù)庫和8小時緩存中。（5）對外接口，供客戶端調(diào)用。2、在本試驗(yàn)中使用的數(shù)據(jù)清洗策略包括3個分別是：（1）平均值策略：用一分鐘產(chǎn)生數(shù)據(jù)的平均值取代這一分鐘產(chǎn)生的所有數(shù)據(jù)，這樣可以在不影響數(shù)據(jù)的大致走勢下減少數(shù)據(jù)冗余減輕數(shù)據(jù)庫的壓力。（2）滑動窗口策略：定義一個固定大小的時間窗口，窗口每次滑動一個讀取周期。如果窗口內(nèi)的標(biāo)簽Tag.次數(shù)大于預(yù)先設(shè)定的最小閾值Min，即認(rèn)為標(biāo)簽Tag.為非噪聲標(biāo)簽，則該窗口就不會輸出標(biāo)簽Tag.?；瑒哟翱诓呗钥梢越o數(shù)據(jù)去噪。（3）移動平均值策略：用該數(shù)據(jù)的前幾次數(shù)據(jù)和后幾次數(shù)據(jù)加上該數(shù)據(jù)本身的平均值去取代該數(shù)據(jù)的值，這樣就消除數(shù)據(jù)的隨機(jī)性。3.5實(shí)驗(yàn)步驟一、新建項(xiàng)目打開VisualStudio2010,選擇C#語言，新建一個類庫程序，修改工程名。如下圖所示。

二、數(shù)據(jù)庫準(zhǔn)備工作因?yàn)橹虚g件要對采集的數(shù)據(jù)進(jìn)行初始化工作而且要把處理之后的數(shù)據(jù)轉(zhuǎn)存到數(shù)據(jù)庫中，所以必須對數(shù)據(jù)庫進(jìn)行操作。這里我們專門新建一個項(xiàng)目來對數(shù)據(jù)庫進(jìn)行操作。第一步:新建項(xiàng)目，右擊解決方案，添加新建項(xiàng)目，命名為：GetDataAccess。第二步：添加SQLHelper的項(xiàng)，這個類的作用是通過存儲過程來對數(shù)據(jù)庫進(jìn)行操作，這里就不對這個類進(jìn)行詳解。首先將老師給的源碼里面的SQLHelper.cs復(fù)制到我們剛剛新4|啟睥決宅2僅個項(xiàng)目）~[etDataAccess|圈生成?LProperty重新生成【日由引用=|DBOperation.csliddleWsre計(jì)算代碼建星值?項(xiàng)目依賴項(xiàng)⑶小項(xiàng)目生成順序m，建的GetDataAccess文件夾下面，然后在VS解決方案中添加現(xiàn)有項(xiàng)，如下圖所示，找到SQLHeleper并添加。新部迎...Ctrl+Shift+A西]新部迎...Ctrl+Shift+A西]現(xiàn)有項(xiàng)回...Shift-FAlt-FA囤Windows窗體心…用戶控件二3添加引用舊...添加S務(wù)引用⑶...設(shè)為啟動項(xiàng)目叫Access項(xiàng)目屈性第三步：新建一個名為DBOperation的類，加入下列代碼，其中連接字符串中的Server一項(xiàng)的IP為數(shù)據(jù)庫所在IP，password一項(xiàng)為數(shù)據(jù)庫的密碼，這兩項(xiàng)應(yīng)該依現(xiàn)實(shí)做相應(yīng)修改。publicclassDBOperation(publicstringconnect="Server=;InitialCatalog=IOTDataBase;uid=sa;password=310;ConnectTimeout=500”;#region連接數(shù)據(jù)庫///<summary>///獲取數(shù)據(jù)庫的連接///</summary>///<returns></returns>privateSqlConnectiongetConnection()(try(stringconnectionString=GetConfigFromXml.DataBaseConnectionString;SqlConnectionconn=newSqlConnection(connectionString);if(conn.State!=ConnectionState.Open)(conn.Open();}returnconn;}catch(Exceptionee)(throwee;}}#endregion///<summary>///獲取數(shù)據(jù)庫中表T_TAG_INDEX中的所有字段///</summary>///<returns></returns>publicDataSetgetTAG_INDEXAllValue()(stringconn=GetConfigFromXml.DataBaseConnectionString;DataSetds=SqlHelper.ExecuteDataset(conn,"zc_SelectTagIndexValue");returnds;}第四步：在數(shù)據(jù)庫中添加相應(yīng)的存儲過程，腳本語言如下。SETANSI_NULLSONGOSETQUOTED_IDENTIFIERONGOCREATEPROCEDURE[dbo].[zc_SelectTagIndexValue]ASBEGIN--SETNOCOUNTONaddedtopreventextraresultsetsfrom--interferingwithSELECTstatements.SETNOCOUNTON;--InsertstatementsforprocedurehereSELECTTAG_ID,TAG_NAME,TYPE_ID,UNIT,MIN,MAX,REMARK,IS_ALARM,SAVE_RATE,DEFAULT_VALfromT_TAG_INDEXEND三、具體編碼（1）新建TagObject類來封裝中間件獲取的數(shù)據(jù)信息。首先右擊項(xiàng)目，找到添加中的新建項(xiàng)，將類名修改為TagObjecto(2)編寫TagObject首先寫入字段屬性，對應(yīng)傳感器或者讀卡器產(chǎn)生數(shù)據(jù)。privatestringtag_id;〃讀卡器ID，傳感器IDprivatestringtag_name;privateinttype_id;//類型編號privatestringunit;〃計(jì)量單位privatestringdefault_value;//默認(rèn)值privatestringmax;//最大值privatestringmin;//最小值privateintrate;//采集頻率privatestringremarks;//說明privateboolis_alarm;//是否報(bào)警privateDateTimegetValueDateTime;//獲取值得時間privateintquality;//質(zhì)量碼privateobjecttagValue;//傳感器所監(jiān)控到的值或者讀卡器讀到的ID給字段屬性添加Set和Get方法，下表只做示例，其余屬性相同處理。publicstringTAG_ID(get(returntag_id;}set(this.tag_id=value;添加必要的構(gòu)造數(shù)和方法。publicTagObject(stringtag_id,stringtag_name)(this.tag_id=tag_id;this.tag_name=tag_name;}publicTagObject(stringtag_id,objecttag_value,DateTimeget_time,intquality)(this.tag_id=tag_id;this.tagValue=tag_value;this.getValueDateTime=get_time;this.quality=quality;}publicoverridestringToString()(DateTimetime=this.GETVALUEDATETIME;if(tagValue!=null)(return(this.tag_name+"-"+time.ToLongTimeString()+":"+this.tagValue.ToString());}else(return(this.tag_name+"-"+time.ToLongTimeString());}}（3）按上述方法依次新建RTDataRemoteObject類（中間件類，負(fù)責(zé)數(shù)據(jù)采集，清洗，及轉(zhuǎn)存），TagCacheL1類（一級緩存，保存所有傳感器讀卡器的實(shí)時值），CacheTagList類（8小時緩存，保存服務(wù)器運(yùn)行8小時的值），DateObject類（保存上次采集時間和采集頻率的類，用途是控制采集頻率）。下面一一詳細(xì)介紹這些類的具體編碼。DateObject類publicclassDateObject（publicintcol_rate;//最采集頻率publicDateTimeget_time;//采集時的時間publicDateObject（intcol_rate,DateTimeget_time）this.col_rate=col_rate;this.get_time=get_time;TagCacheLl類，是中間件的一級緩存。publicclassTagCacheLl(privatestaticTagCacheL1myInstance;//獲取自己的一個實(shí)例publicHashtabletagValueHashtable=Hashtable.Synchronized(newHashtable());//保存實(shí)時讀到的鍵值對///<summary>///返回一個CacheL1對象///</summary>///<returns></returns>publicstaticTagCacheL1GetInstance()(if(myInstance==null)(myInstance=newTagCacheL1();}returnmyInstance;}}CacheTagList類，是8小時緩存publicclassCacheTagList(///<summary>///存放的是tag_name以及對應(yīng)的包含平均值的一個TagObject對象///</summary>publicHashtablecurrentTagMinuteAverageHt=newHashtable();///<summary>///存放的是tag_name以及對應(yīng)的包含多組平均值的Array[]數(shù)組///</summary>publicHashtabletagNameToValueArraysHT=null;publicintip_position=0;///<summary>///初始化tagNameToValueArraysHT///</summary>///<paramname="tagName">包含tag_name的數(shù)組</param>///<returns></returns>publicboolIntilize8HDB(string[]tagName)(boolflag=false;this.tagNameToValueArraysHT=newHashtable();for(inti=0;i<tagName.Length;i++)(object[]objArray=newobject[480];this.tagNameToValueArraysHT.Add(tagName[i],objArray);flag=true;}returnflag;}///<summary>///將currentTagMinuteAverageHt中存儲的平均值///存入到tagNameToValueArraysHT中的array[]中///</summary>publicvoidAddMinuteAverageValueToHt()(try(object[]objArray;if(this.ip_position>=0x1df)//0x1df=479(this.ip_position=0;}else(this.ip_position++;}if(this.currentTagMinuteAverageHt!=null)(IDictionaryEnumeratorenumerator=this.currentTagMinuteAverageHt.GetEnumerator();while(enumerator.MoveNext())(stringkey=(string)enumerator.Key;〃當(dāng)currentTagMinuteAverageHt和tagNameToValueArraysHT包含相同key時if(this.tagNameToValueArraysHT.ContainsKey(key)){//currentTagMinuteAverageHt中的值是TagObject對象TagObjectobj2=(TagObject)enumerator.Value;//tagNameToValueArraysHT中的值是一個Object[]數(shù)組〃數(shù)組中存放的是一個TagObject的tagValue值objArray=(object[])this.tagNameToValueArraysHT[key];if(obj2!=null)(objArray[this.ip_position]=obj2;this.tagNameToValueArraysHT[key]=objArray;}}}}}catch(Exceptione)(LogWriter.LogE("實(shí)時數(shù)據(jù)采集-分鐘平均值緩存",e.Message);}}RTDataRemoteObject類，是最重要的類，其負(fù)責(zé)數(shù)據(jù)的采集、清洗與轉(zhuǎn)存，還提供對外的數(shù)據(jù)接口。我們分功能對其進(jìn)行展示。&導(dǎo)入包及定義變量首先引用包usingSystem.Collections;usingSystem.Windows.Forms;usingSystem.Threading;usingSystem.Diagnostics;usingGetDataAcess;其中System.Windows.Forms的使用需要添加.NET引用，在解決方案資源管理器中，右擊引用，如下圖所示。霧突方段器▼畢xhIS.?.囹MiddlewareWProperties畫'J添加引用回…硝匚瀚]澳務(wù)弓!用㈤…碧GetConfigFromXmI.cs盟LogWriter.es費(fèi)OPC.cs費(fèi)RTDataRemoteOLJect.cs*TagCacheLl.cs購TagObjsct.csOPCdotNETLib然后，在添加引用窗口中找到.NET選擇卡，添加其中的System.Windows.Form.如下所示。8添加弓1用]NET1項(xiàng)目|瞄［知篩選為：，METFramework4沮件名稱版本運(yùn)行時路徑，System.Web.Routing4.0.0.Dv4.0.30319C:\ProgramFiles(x&6)\System.Web.Servicesv4.0.30319C:\ProgramFiles(x&6)\Syst&tn.Windows.Fortns.D...v4.0.30319C:\ProgramFiles(x&6]\System.Windciws.Forms.D...V4.0.30319CAProgramFiles〔乂86八]Sy戒em,Wind□ws.Form54.ao.oV4.O.3O319C:\PrcigramSystem.Windciws.TnputM...v4.0.30319C:\PrograiTiFiles(k&6]\System.Windows.Present...4.D.D.0v4.0.30319C:\ProgramFiles(x860\System.Workflow.Activitiesv4.0.30319C:\ProgramFiles(x86)\System.Workflow.Compo...4.D.0.Dv4.0.30319C:\ProgramFiles(x86)\System.Workflow.Runtimev4.O.3O319C:\PrograrnFilas(x36]\SvstmtTL'WcirkRciwSiervicES叮4.0.D.Orrrv4.0.30319C:\PrciciramFil*fx&61\TL誦定前有類似的我們還需要引用GetDataAcess.dll,如下圖所示。最后我們定義必要的變量；#region定義變量DBOperationdb=newDBOperation();〃一級緩存publicstaticTagCacheLltagCacheLl=newTagCacheL1();//八小時緩存publicstaticCacheTagListCache8HTagList=newCacheTagList();//8小時緩存是否啟動publicboolis8HDBInitilized=false;//用于保存從OPC中取出tagid與tagobject的鍵值對publicstaticHashtableOPCtagCacheLHashtable=Hashtable.Synchronized(newHashtable());//保存tag的tag_name與tag_id鍵值對publicstaticHashtabletagHashtable=Hashtable.Synchronized(newHashtable());//OPCtag_id字符串，用于構(gòu)造Hashtable的鍵值privatestring[]strOPCTagId=null;//保存DateObject與Tag_id的鍵值對，用于采集頻率的計(jì)算publicstaticHashtableDateHashtable=Hashtable.Synchronized(newHashtable());//保存tag每分鐘的值的hashtable(用于分鐘平均值計(jì)算)publicstaticHashtableOPCTagMinuteValuesHashtable=Hashtable.Synchronized(newHashtable());//保存tag標(biāo)簽的最大值最小值publicstaticHashtableMINorMAXHashtable=Hashtable.Synchronized(newHashtable());//保存計(jì)算后的移動平均值的實(shí)時值publicstaticHashtablemovingAverageHt=Hashtable.Synchronized(newHashtable());//保存計(jì)算前移動平均值需要的數(shù)publicstaticHashtablemovingAverageListHt=Hashtable.Synchronized(newHashtable());//移動平均值窗口大小staticintWSIZE=5;//保存滑動窗口去噪策略最小閾值(不同的標(biāo)簽會有不同的閾值)publicstaticHashtableMINHashtable=Hashtable.Synchronized(newHashtable());〃是否啟動移動平均值策略清洗數(shù)據(jù)publicboolisStartMovingAverage=false;//平均值計(jì)算線程privateThreaddataComputeThread=null;privateSystem.Threading.TimerdataComputeTimer=null;//OPC數(shù)據(jù)采集線程privateThreadOPCcurrentDataHarvestThread=null;privateSystem.Threading.TimerOPCDataHarvestTimer=null;publicHashtableOPCHashTable=newHashtable();//privateOPCOpc=newOPC();〃線程等待時間初始化函數(shù)，因?yàn)槲覀儾荒芤蠼⒅虚g件對象的時候，就要讓它工作；#region啟動函數(shù)publicvoidInitRemoteObject()(//初始化一級緩存this.InitCacheLFromDB();//初始化8小時緩存this.InitCacheTagList();//采集。PCServer產(chǎn)生的實(shí)時數(shù)據(jù)this.OPCDataHarvestThread();//平均值計(jì)算線程包括轉(zhuǎn)存數(shù)據(jù)this.DtaComputeThread();//this.DataHarvestThread();}#endregionC:準(zhǔn)備工作，初始化一級緩存，作用是為采集，處理數(shù)據(jù)做初始化工作;#region初始化一級緩存publicvoidInitCacheLFromDB()(try(DataSetds=db.getTAG_INDEXAllValue();DataTabledt=ds.Tables[0];intcount=dt.Rows.Count;this.strOPCTagId=newstring[count];for(inti=0;i<count;i++)(string[]MAXMIN=newstring[2];//保存最大最小值stringtag_id=dt.Rows[i]["TAG_ID"].ToString().Trim();stringtag_name=dt.Rows[i]["TAG_NAME"].ToString();intrate=int.Parse(dt.Rows[i]["SAVE_RATE"].ToString());DateObjectdateobj=newDateObject(rate,DateTime.Now);DateHashtable.Add(tag_id,dateobj);this.strOPCTagId[i]=tag_id;TagObjectobj2=newTagObject(tag_id,tag_name);obj2.RATE=rate;if(dt.Rows[i]["TYPE_ID"]!=null)(obj2.TYPE_ID=int.Parse(dt.Rows[i]["TYPE_ID"].ToString());}if(dt.Rows[i]["UNIT"]!=null)(obj2.UNIT=dt.Rows[i]["UNIT"].ToString();}if(dt.Rows[i]["DEFAULT_VAL"]!=null)(obj2.TAGVALUE=dt.Rows[i]["DEFAULT_VAL"].ToString();}if(dt.Rows[i]["MIN"]!=null)(obj2.MIN=dt.Rows[i]["MIN"].ToString();MAXMIN[1]=dt.Rows[i]["MIN"].ToString();}if(dt.Rows[i]["MAX"]!=null)(obj2.MAX=dt.Rows[i]["MAX"].ToString();MAXMIN[0]=dt.Rows[i]["MAX"].ToString();}obj2.RATE=rate;if(dt.Rows[i]["REMARK"]!=null)(obj2.REMARKS=dt.Rows[i]["REMARK"].ToString();}if(dt.Rows[i]["IS_ALARM"]!=null)(obj2.IS_ALARM=bool.Parse(dt.Rows[i]["IS_ALARM"].ToString());}obj2.GETVALUEDATETIME=DateTime.Now;obj2.QUALITY=1;//tagCacheLHashtable.Add(tag_id,obj2);OPCtagCacheLHashtable.Add(tag_id,obj2);MINorMAXHashtable.Add(tag_id,MAXMIN);tagHashtable.Add(tag_id,tag_name);movingAverageListHt.Add(tag_id,newArrayList());tagCacheL1.tagValueHashtable=OPCtagCacheLHashtable;inttimes=60000/rate;MINHashtable.Add(tag_id,times);}}catch(Exceptione)MessageBox.Show("Error:初始化一級緩存出錯”+e.Message);throw(e);JD：準(zhǔn)備工作，初始化8小時緩存；#region初始化八小時緩存///<summary>///根據(jù)傳感器TAG初始化八小時緩存///</summary>///<returns></returns>publicvoidInitCacheTagList()(try(Cache8HTagList.Intilize8HDB(this.strOPCTagId);this.is8HDBInitilized=true;}catch(Exceptione)(MessageBox.Show(”實(shí)時數(shù)據(jù)采集-初始化八小時緩存:"+e.Message);}}#endregionE：數(shù)據(jù)采集工作，采用線程每隔一秒中執(zhí)行一次采集函數(shù)，采集一組數(shù)據(jù)到緩存中。大家注意代碼標(biāo)紅的那一段，這一段的作用就是設(shè)置實(shí)時預(yù)警，當(dāng)數(shù)據(jù)大于最大值，小于最小值時，報(bào)警標(biāo)識標(biāo)記為True；#regionOPCServer實(shí)時數(shù)據(jù)采集publicvoidOPCDataHarvestThread()(ThreadStartstart=newThreadStart(this.OPCDataHarvestService);this.OPCcurrentDataHarvestThread=newThread(start);this.OPCcurrentDataHarvestThread.Start();}protectedvoidOPCDataHarvestService()(intdeTime=this.wait;〃延后時間TimerCallbackCallback=newTimerCallback(this.OPCCurrentDataHarvest);this.OPCDataHarvestTimer=newSystem.Threading.Timer(Callback,null,deTime,1000);publicvoidOPCCurrentDataHarvest(objectstate)(ArrayListtagIds=newArrayList();try(Hashtableht=newHashtable(DateHashtable);IDictionaryEnumeratorenumerator=ht.GetEnumerator();if(enumerator!=null)(while(enumerator.MoveNext())(stringkey=enumerator.Key.ToString();DateObjectdateObject=(DateObject)enumerator.Value;if(DateTime.Now>=dateObject.get_time.AddMilliseconds(dateObject.col_rate))(tagIds.Add(key);dateObject.get_time=DateTime.Now;DateHashtable[key]=dateObject;}}}}catch(Exceptione)(MessageBox.Show("Error:OPC采集實(shí)時數(shù)據(jù)出錯，出錯段為給tag刷新數(shù)據(jù):"+e.Message);throw(e);}try(if(tagIds!=null)(for(inti=0;i<tagIds.Count;i++)(stringtagId=tagIds[i].ToString();if(OPCtagCacheLHashtable.ContainsKey(tagId))(TagObjecttagObject=(TagObject)OPCtagCacheLHashtable[tagId];if(tagObject!=null)(try(if(OPCHashTable[tagId]!=null)(tagObject.TAGVALUE=((TagObject)OPCHashTable[tagId]).TAGVALUE;tagObject.GETVALUEDATETIME=((TagObject)OPCHashTable[tagId]).GETVALUEDATETIME;tagObject.QUALITY=((TagObject)OPCHashTable[tagId]).QUALITY;if(MINorMAXHashtable[tagObject.TAG_ID]!=null)(tagObject.MAX=((string[])MINorMAXHashtable[tagObject.TAG_ID])[0];tagObject.MIN=((string[])MINorMAXHashtable[tagObject.TAG_ID])[1];}doublevalue=Convert.ToDouble(tagObject.TAGVALUE);doublemin=Convert.ToDouble(tagObject.MIN);doublemax=Convert.ToDouble(tagObject.MAX);if(value<max&&value>=min)(tagObject.IS_ALARM=false;〃當(dāng)標(biāo)簽是溫度時，即使是最小值也需要報(bào)警if(tagObject.TAG_ID.Contains("TP")&&value==min)(tagObject.IS_ALARM=true;}}else(tagObject.IS_ALARM=true;}if(tagCacheL1.tagValueHashtable.ContainsKey(tagObject.TAG_ID))(tagCacheL1.tagValueHashtable[tagObject.TAG_ID]=tagObject;//將數(shù)據(jù)轉(zhuǎn)存到OPCTagMinuteValuesHashtable，此哈希表用于求每分鐘的平均值elsetagObject.TAGVALUE=-1;tagObject.GETVALUEDATETIME=DateTime.Now;「…=0;if(isStartMovingAverage)”"catch(Exceptione)MessageBox.Show("Error:OPC采集實(shí)時數(shù)據(jù)出錯，出錯段為給tag刷新數(shù)據(jù):"+e.Message);throw(e);}}}}}catch(Exceptione)(MessageBox.Show(”實(shí)時數(shù)據(jù)采集-OPC采集實(shí)時數(shù)據(jù)出錯:"+e.Message);throw(e);}}publicboolAddTagValueToTagList(TagObjecttagObject)(boolflag=true;try(//這些TagObject對像必須是在Cache中的if(tagObject!=null)(ArrayListlist;if(OPCTagMinuteValuesHashtable.ContainsKey(tagObject.TAG_ID))(list=(ArrayList)OPCTagMinuteValuesHashtable[tagObject.TAG_ID];list.Add(tagObject.TAGVALUE);OPCTagMinuteValuesHashtable[tagObject.TAG_ID]=list;}else(list=newArrayList();list.Add(tagObject.TAGVALUE);OPCTagMinuteValuesHashtable.Add(tagObject.TAG_ID,list);}}}catch(Exceptione)(flag=false;MessageBox.Show(”實(shí)時數(shù)據(jù)采集-存入一分鐘緩存:"+e.Message);}returnflag;}#endregionF：移動平均值策略#region計(jì)算移動平均值publicvoidComputeMovingAverage(TagObjecttag)(doublesecond=-(WSIZE-1)/2;doublesum=0;doubleaverageValue=0;TagObjectaverageTag;if(movingAverageListHt.ContainsKey(tag.TAG_ID))(ArrayListmovingAverageList=(ArrayList)movingAverageListHt[tag.TAG_ID];if(movingAverageList.Count<WSIZE-1)(movingAverageList.Add(tag.TAGVALUE);DateTimedtime=tag.GETVALUEDATETIME.AddSeconds(second);averageTag=newTagObject(tag.TAG_ID,-1,dtime,0);}else(movingAverageList.Add(tag.TAGVALUE);for(inti=0;i<movingAverageList.Count;i++)(doubled=Convert.ToDouble(movingAverageList[i]);sum+=d;}//decimal.Round(Convert.ToDecimal(floatPart),3);averageValue=(double)decimal.Round(Convert.ToDecimal(sum/WSIZE),3);DateTimedtime=tag.GETVALUEDATETIME.AddSeconds(second);averageTag=newTagObject(tag.TAG_ID,averageValue,dtime,1);movingAverageList.RemoveAt(0);}if(movingAverageHt.ContainsKey(tag.TAG_ID))(movingAverageHt[tag.TAG_ID]=averageTag;}else(movingAverageHt.Add(tag.TAG_ID,averageTag);}}}#endregionG:平均值策略和滑動窗口策略，平均值策略即求1分鐘的平均值，減少了存入數(shù)據(jù)庫數(shù)據(jù)的冗余，又不影響數(shù)據(jù)的走勢。而滑動窗口策略是為了給數(shù)據(jù)去噪，去掉那些極可能是干擾的數(shù)據(jù)。#region一分鐘數(shù)據(jù)平均值計(jì)算線程publicvoidDtaComputeThread()(ThreadStartstart=newThreadStart(this.dataComputeService);this.dataComputeThread=newThread(start);this.dataComputeThread.Start();protectedvoiddataComputeService()(intdelay=0x7530;//30秒intInterval=0xea60;//60秒TimerCallbackCallback=newTimerCallback(this.MinuteDataAverageCompute);this.dataComputeTimer=newSystem.Threading.Timer(Callback,null,delay,Interval);}publicvoidMinuteDataAverageCompute(objectstate)(HashtabletagMinuteValuesHashtable=newHashtable(OPCTagMinuteValuesHashtable);OPCTagMinuteValuesHashtable.Clear();this.ComputingAllTagValuesMinuteAverage(tagMinuteValuesHashtable);Cache8HTagList.AddMinuteAverageValueToHt();}#region計(jì)算每分鐘數(shù)據(jù)的平均值///<summary>///tagMinuteValuesHashtable中的鍵為tag_name值為list里面存放的是tag_Value///</summary>///<returns></returns>publicboolComputingAllTagValuesMinuteAverage(HashtabletagMinuteValuesHashtable)(boolflag=false;IDictionaryEnumeratorenumerator=null;if(tagMinuteValuesHashtable!=null)(enumerator=tagMinuteValuesHashtable.GetEnumerator();}try(if(enumerator==null)(returnflag;}while(enumerator.MoveNext())(〃針對每個tag_namestringkey=(string)enumerator.Key;ArrayListlist=(ArrayList)enumerator.Value;〃針對每個Tagname求取if(list!=null)(lEnumeratorenumerator2=list.GetEnumerator();intcount=list.Count;//滑動窗口去噪,如果采集的數(shù)量小于設(shè)定的最小閾值說明是噪聲，應(yīng)該去掉inttimes=(int)MINHashtable[key];if(times!=null)(if(count<times/3)(continue;}}doublenum1=0f;if(count>0)(doublenum2=0f;//if(sensorTagHashtable.ContainsKey(key))//tag_id=sensorTagHashtable[key].ToString();while(enumerator2.MoveNext())(if(enumerator2.Current!=null)(//求和num2+=Convert.ToDouble(enumerator2.Current);}}num1=num2/((float)count);num1=(double)decimal.Round(Convert.ToDecimal(num1),3);}TagObjectcurrent=newTagObject(key,num1,DateTime.Now,1);db.InsertMinuteAverage(key,DateTime.Now,num1,1);if(Cache8HTagList.currentTagMinuteAverageHt.ContainsKey(key))(Cache8HTagList.currentTagMinuteAverageHt[key]=current;}else(Cache8HTagList.currentTagMinuteAverageHt.Add(key,current);}flag=true;}else(flag=false;}}}catch(Exceptione)(flag=false;MessageBox.Show("實(shí)時數(shù)據(jù)采集-平均值計(jì)算”+e.Message);}returnflag;}#endregion#region清空保存?zhèn)鞲衅黝愐环昼姅?shù)據(jù)的HashTable///<summary>///清空保存?zhèn)鞲衅黝愐环昼姅?shù)據(jù)的HashTable///</summary>///<returns></returns>publicboolRemoveAllMinuteTagValues()(boolflag=true;if(OPCTagMinuteValuesHashtable!=null)(IDictionaryEnumeratorenumerator=OPCTagMinuteValuesHashtable.GetEnumerator();try(while(enumerator.MoveNext())(stringkey=(string)enumerator.Key;((ArrayList)OPCTagMinuteValuesHashtable[key]).Clear();}}catch(InvalidOperationExceptione)(flag=false;MessageBox.Show("實(shí)時數(shù)據(jù)采集-清空傳感器一分鐘數(shù)據(jù)"+e.Message);}}returnflag;}#endregion#endregionH:對外接口，供應(yīng)用軟件調(diào)用。#region實(shí)時數(shù)據(jù)訪問接口///<summary>///獲取TAG最新讀取到的值///</summary>///<returns>TagObject</returns>publicTagObjectgetRealtimeTagValue(stringtagId)(TagObjectobj=null;if(tagCacheL1.tagValueHashtable.ContainsKey(tagId))(TagObjectobj2=(TagObject)tagCacheL1.tagValueHashtable[tagId];if(obj2!=null)(obj=obj2.Clone();}}returnobj;}///<summary>///啟動移動平均值策略///</summary>publicvoidStartMovingAverage()(this.isStartMovingAverage=true;return;}///<summary>///停止移動平均值策略///</summary>publicvoidStopMovingAverage()(this.isStartMovingAverage=false;return;}///<summary>///獲取移動平均值的實(shí)時值///</summary>///<paramname="tagId"></param>///<returns></returns>publicTagObjectgetMovingAverageTagValue(stringtagId)(if(isStartMovingAverage==true)(TagObjectobj=null;if(movingAverageHt.ContainsKey(tagId))(TagObjectobj2=(TagObject)movingAverageHt[tagId];if(obj2!=null)(obj=obj2.Clone();}}returnobj;}else(returnnull;}}///<summary>///根據(jù)輸入的【。數(shù)組獲取對應(yīng)的實(shí)時值鏈表///</summary>///<paramname="tagId"></param>///<returns></returns>publicArrayListgetTagValueList(string[]tagId)(ArrayListtagList=newArrayList();TagObjectobj=null;inti=tagId.Length;for(intj=0;j<i;j++)(if(tagCacheL1.tagValueHashtable.ContainsKey(tagId[j]))(TagObjectobj2=(TagObject)tagCacheL1.tagValueHashtable[tagId[j]];if(obj2!=null)(obj=obj2.Clone();}}tagList.Add(obj);}returntagList;}///<summary>Ill獲取OPCServer的最新值///<lsummary>Ill<paramname="ht"><lparam>Ill<paramname="secret"><lparam>publicvoidSetOPCHashtable(Hashtableht,stringsecret)(if(ht!=null&&secret=="310”)(OPCHashTable=ht;}return;}lll<summary>lll獲取傳感器最近八個小時的平均值，值的個數(shù)由arrayLength決定lll<lsummary>lll<returns>object<lreturns>publicobject[]getTagObjectRange(stringtagId,intarrayLength)(object[]objArray=null;intindex=Cache8HTagList.ip_position;if(Cache8HTagList.tagNameToValueArraysHT.ContainsKey(tagId))(objArray=newobject[arrayLength];object[]objArray2=(object[])Cache8HTagList.tagNameToValueArraysHT[tagId];for(inti=0;i<objArray.Length;i++)(if(index<0)(index=0x1df;}objArray[i]=null;if(objArray2[index]!=null)(objArray[i]=((TagObject)objArray2[index]).TAGVALUE;}index--;}}returnobjArray;}publicArrayListgetTagObjectRangeBytime(stringtagId,DateTimetimel,DateTimetime2)(ArrayListlist=newArrayList();intindex=Cache8HTagList.ipposition+1;if(Cache8HTagList.tagNameToValueArraysHT.ContainsKey(tagId))(object[]objArray=(object[])Cache8HTagList.tagNameToValueArraysHT[tagId];for(inti=0;i<480;i++)(if(index>0x1df)(index=0;}if(objArray[index]!=null)(TagObjecttag=(TagObject)objArray[index];if(tag.GETVALUEDATETIME>=time1&&tag.GETVALUEDATETIME<=time2)(list.Add(tag);}}index++;}}returnlist;}publicHashtablegetHash()(returntagCacheL1.tagValueHashtable;}#endregion至此，中間件的編寫就初步結(jié)束了。但是此時的中間件是不能獨(dú)立運(yùn)行的，因?yàn)槠溥€需要與OPCClient連接起來去OPCServer中取數(shù)據(jù)。我們打開前一個實(shí)驗(yàn)完成的IOTServer項(xiàng)目，將我們剛剛完成的中間件MiddleWare.dll添加到該項(xiàng)目的引用中，展開Mai