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

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

二、數(shù)據(jù)庫準備工作因為中間件要對采集的數(shù)據(jù)進行初始化工作而且要把處理之后的數(shù)據(jù)轉存到數(shù)據(jù)庫中，所以必須對數(shù)據(jù)庫進行操作。這里我們專門新建一個項目來對數(shù)據(jù)庫進行操作。第一步:新建項目，右擊解決方案，添加新建項目，命名為：GetDataAccess。第二步：添加SQLHelper的項，這個類的作用是通過存儲過程來對數(shù)據(jù)庫進行操作，這里就不對這個類進行詳解。首先將老師給的源碼里面的SQLHelper.cs復制到我們剛剛新4|啟睥決宅2僅個項目）~[etDataAccess|圈生成?LProperty重新生成【日由引用=|DBOperation.csliddleWsre計算代碼建星值?項目依賴項⑶小項目生成順序m，建的GetDataAccess文件夾下面，然后在VS解決方案中添加現(xiàn)有項，如下圖所示，找到SQLHeleper并添加。新部迎...Ctrl+Shift+A西]新部迎...Ctrl+Shift+A西]現(xiàn)有項回...Shift-FAlt-FA囤Windows窗體心…用戶控件二3添加引用舊...添加S務引用⑶...設為啟動項目叫Access項目屈性第三步：新建一個名為DBOperation的類，加入下列代碼，其中連接字符串中的Server一項的IP為數(shù)據(jù)庫所在IP，password一項為數(shù)據(jù)庫的密碼，這兩項應該依現(xiàn)實做相應修改。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ù)庫中添加相應的存儲過程，腳本語言如下。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ù)信息。首先右擊項目，找到添加中的新建項，將類名修改為TagObjecto(2)編寫TagObject首先寫入字段屬性，對應傳感器或者讀卡器產生數(shù)據(jù)。privatestringtag_id;〃讀卡器ID，傳感器IDprivatestringtag_name;privateinttype_id;//類型編號privatestringunit;〃計量單位privatestringdefault_value;//默認值privatestringmax;//最大值privatestringmin;//最小值privateintrate;//采集頻率privatestringremarks;//說明privateboolis_alarm;//是否報警privateDateTimegetValueDateTime;//獲取值得時間privateintquality;//質量碼privateobjecttagValue;//傳感器所監(jiān)控到的值或者讀卡器讀到的ID給字段屬性添加Set和Get方法，下表只做示例，其余屬性相同處理。publicstringTAG_ID(get(returntag_id;}set(this.tag_id=value;添加必要的構造數(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類（中間件類，負責數(shù)據(jù)采集，清洗，及轉存），TagCacheL1類（一級緩存，保存所有傳感器讀卡器的實時值），CacheTagList類（8小時緩存，保存服務器運行8小時的值），DateObject類（保存上次采集時間和采集頻率的類，用途是控制采集頻率）。下面一一詳細介紹這些類的具體編碼。DateObject類publicclassDateObject（publicintcol_rate;//最采集頻率publicDateTimeget_time;//采集時的時間publicDateObject（intcol_rate,DateTimeget_time）this.col_rate=col_rate;this.get_time=get_time;TagCacheLl類，是中間件的一級緩存。publicclassTagCacheLl(privatestaticTagCacheL1myInstance;//獲取自己的一個實例publicHashtabletagValueHashtable=Hashtable.Synchronized(newHashtable());//保存實時讀到的鍵值對///<summary>///返回一個CacheL1對象///</summary>///<returns></returns>publicstaticTagCacheL1GetInstance()(if(myInstance==null)(myInstance=newTagCacheL1();}returnmyInstance;}}CacheTagList類，是8小時緩存publicclassCacheTagList(///<summary>///存放的是tag_name以及對應的包含平均值的一個TagObject對象///</summary>publicHashtablecurrentTagMinuteAverageHt=newHashtable();///<summary>///存放的是tag_name以及對應的包含多組平均值的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;〃當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ù)據(jù)采集-分鐘平均值緩存",e.Message);}}RTDataRemoteObject類，是最重要的類，其負責數(shù)據(jù)的采集、清洗與轉存，還提供對外的數(shù)據(jù)接口。我們分功能對其進行展示。&導入包及定義變量首先引用包usingSystem.Collections;usingSystem.Windows.Forms;usingSystem.Threading;usingSystem.Diagnostics;usingGetDataAcess;其中System.Windows.Forms的使用需要添加.NET引用，在解決方案資源管理器中，右擊引用，如下圖所示。霧突方段器▼畢xhIS.?.囹MiddlewareWProperties畫'J添加引用回…硝匚瀚]澳務弓!用㈤…碧GetConfigFromXmI.cs盟LogWriter.es費OPC.cs費RTDataRemoteOLJect.cs*TagCacheLl.cs購TagObjsct.csOPCdotNETLib然后，在添加引用窗口中找到.NET選擇卡，添加其中的System.Windows.Form.如下所示。8添加弓1用]NET1項目|瞄［知篩選為：，METFramework4沮件名稱版本運行時路徑，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字符串，用于構造Hashtable的鍵值privatestring[]strOPCTagId=null;//保存DateObject與Tag_id的鍵值對，用于采集頻率的計算publicstaticHashtableDateHashtable=Hashtable.Synchronized(newHashtable());//保存tag每分鐘的值的hashtable(用于分鐘平均值計算)publicstaticHashtableOPCTagMinuteValuesHashtable=Hashtable.Synchronized(newHashtable());//保存tag標簽的最大值最小值publicstaticHashtableMINorMAXHashtable=Hashtable.Synchronized(newHashtable());//保存計算后的移動平均值的實時值publicstaticHashtablemovingAverageHt=Hashtable.Synchronized(newHashtable());//保存計算前移動平均值需要的數(shù)publicstaticHashtablemovingAverageListHt=Hashtable.Synchronized(newHashtable());//移動平均值窗口大小staticintWSIZE=5;//保存滑動窗口去噪策略最小閾值(不同的標簽會有不同的閾值)publicstaticHashtableMINHashtable=Hashtable.Synchronized(newHashtable());〃是否啟動移動平均值策略清洗數(shù)據(jù)publicboolisStartMovingAverage=false;//平均值計算線程privateThreaddataComputeThread=null;privateSystem.Threading.TimerdataComputeTimer=null;//OPC數(shù)據(jù)采集線程privateThreadOPCcurrentDataHarvestThread=null;privateSystem.Threading.TimerOPCDataHarvestTimer=null;publicHashtableOPCHashTable=newHashtable();//privateOPCOpc=newOPC();〃線程等待時間初始化函數(shù)，因為我們不能要求建立中間件對象的時候，就要讓它工作；#region啟動函數(shù)publicvoidInitRemoteObject()(//初始化一級緩存this.InitCacheLFromDB();//初始化8小時緩存this.InitCacheTagList();//采集。PCServer產生的實時數(shù)據(jù)this.OPCDataHarvestThread();//平均值計算線程包括轉存數(shù)據(jù)this.DtaComputeThread();//this.DataHarvestThread();}#endregionC:準備工作，初始化一級緩存，作用是為采集，處理數(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：準備工作，初始化8小時緩存；#region初始化八小時緩存///<summary>///根據(jù)傳感器TAG初始化八小時緩存///</summary>///<returns></returns>publicvoidInitCacheTagList()(try(Cache8HTagList.Intilize8HDB(this.strOPCTagId);this.is8HDBInitilized=true;}catch(Exceptione)(MessageBox.Show(”實時數(shù)據(jù)采集-初始化八小時緩存:"+e.Message);}}#endregionE：數(shù)據(jù)采集工作，采用線程每隔一秒中執(zhí)行一次采集函數(shù)，采集一組數(shù)據(jù)到緩存中。大家注意代碼標紅的那一段，這一段的作用就是設置實時預警，當數(shù)據(jù)大于最大值，小于最小值時，報警標識標記為True；#regionOPCServer實時數(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ù)據(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;〃當標簽是溫度時，即使是最小值也需要報警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ù)轉存到OPCTagMinuteValuesHashtable，此哈希表用于求每分鐘的平均值elsetagObject.TAGVALUE=-1;tagObject.GETVALUEDATETIME=DateTime.Now;「…=0;if(isStartMovingAverage)”"catch(Exceptione)MessageBox.Show("Error:OPC采集實時數(shù)據(jù)出錯，出錯段為給tag刷新數(shù)據(jù):"+e.Message);throw(e);}}}}}catch(Exceptione)(MessageBox.Show(”實時數(shù)據(jù)采集-OPC采集實時數(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ù)據(jù)采集-存入一分鐘緩存:"+e.Message);}returnflag;}#endregionF：移動平均值策略#region計算移動平均值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ù)平均值計算線程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計算每分鐘數(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ù)量小于設定的最小閾值說明是噪聲，應該去掉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ù)據(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ù)據(jù)采集-清空傳感器一分鐘數(shù)據(jù)"+e.Message);}}returnflag;}#endregion#endregionH:對外接口，供應用軟件調用。#region實時數(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>///獲取移動平均值的實時值///</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ù)組獲取對應的實時值鏈表///</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至此，中間件的編寫就初步結束了。但是此時的中間件是不能獨立運行的，因為其還需要與OPCClient連接起來去OPCServer中取數(shù)據(jù)。我們打開前一個實驗完成的IOTServer項目，將我們剛剛完成的中間件MiddleWare.dll添加到該項目的引用中，展開Mai