基于新框架的开发多线程框架下开发

《基于新框架的开发多线程框架下开发》由会员分享，可在线阅读，更多相关《基于新框架的开发多线程框架下开发（31页珍藏版）》请在装配图网上搜索。

1、框架下开发基于新框架的开发目录结构billing40/frame/svr/xdrframe/frame_interface.h 框架billing40/app/framefile/frame_file_interface.h文件(输入，输入，事物控制，状态) billing40在products/openboss/billing40用:cvs_co products/openboss/billing40cd products/openboss/billing40cvs_up r ob20_dev开发如批价：class CRatingThread : public CThreadPublic:V

2、irtual int32 init();Virtual int32 destroy();Virtual int32 run(CData *p);virtual int32 exception(CData *p);virtual int32 control(DOMNode & resultDoc, const DOMNode & controlDoc);CThread *create_instanceRating()Return new CRatingThread();l 所有在框架下运行的lib都有一个class，这个class直接或间接继承CThreadl 在这个class里，须实现虚拟函数

3、run。如：对批价来说run的参数p就是一个xdr的指针l Lib里还须提供一个函数，创建这个class的对象，如上面create_instance，框架为每个线程创建一个class的对象。l 每个线程一个CRatingThread对象，在CRatingThread里的数据都是线程安全的l 处理完成的数据，调用send_data(void *p, const int32& iIndex)，把数据送到下一个节点（模块）。参数iIndex，是在当前节点的处理结果要分开输出到不同节点时用。如正确话单iIndex=0, 错单iIndex=1。l Init函数。线程初始化。如读配置文件，连接到MDBse

4、rver等。l 时钟，CTime *g_pTime，每100微秒更新一次。CTime里有多格式的时间，可以直接使用，如批价使用的process_time。CTime的声明见下面l 配置文件采用xml，而且原来有一些配置到数据库表里的配置也移到了配置文件，框架和数据库没有关系了，不再提供otl_connect，也没有了数据库帐号，读取配置项统一用CThread里的read_config 。框架已经对配置文件进行了初始化，各模块不需要再对配置文件做初始化，直接读取即可。框架提供了一些读取配置项的宏，可以方便地读取到配置项。如业务分析的配置，见下面举例。l 多线程框架在名字空间bs_frame里l

5、包含头文件frame_interface.h，编译时使用-lFrameDl Exception函数是异常时的处理，如批价，在异常时，把话单打成错单。l 写日志统一采用CThread:LogAppend，框架可以会把一些日志发到前台。同时，通过这个函数写日志时，框架会自动的写日志的线程所在的流水线和节点的信息加上。这也带来了一个麻烦，写日志时必需有一个CThread对象(业务处理模块都有一个类（A）从CThread派生，在这个类里可以直接调用LogAppend，在其它不是从CThread类派生的类里，如果需要写日志，需要把A的指针传给它，以使这些类也可以写日志。)l 节点级变量，当节点内的所有线

6、程使用相同的变量时，变量可以放到节点里，变量在节点内只有一份，节点级变量只能被设置一次。class CTimepublic: struct timeval m_tmVal; / m_tmVal. tv_sec从1970年到当前时间的秒数, m_tmVal. tv_usec 精确到微秒 struct tm m_tm; int32 m_iYYYYMMDD; /YYYYMMDD int32 m_iYYYYMM; /YYYYMM int32 m_iYYYY; /YYYY int64 m_llYYYYMMDDHH24MISS; /YYYYMMDDHH24MISS char m_szYYYY8; /YYY

7、Y char m_szYYYYMM8; /YYYYMM char m_szYYYYMMDD9; /YYYYMMDD char m_szYYYYMMDDHH24MISS15; /YYYYMMDDHH24MISS char m_szDateTimeMs / YYYY-MM-DD HH24:MI:SS.MS6后面是6位数的微秒 static char *to_stringDateTime(char *szBuf, const time_t& lTime); static char *to_stringDate(char *szBuf, const time_t& lTime); static int

8、64& to_stringDateTime(int64& llDateTime, const time_t& lTime); static int32& to_stringDate(int32& iDate, const time_t& lTime);事物控制事物控制总的指导思想是，对有需要做事物控制的模块，将其处理完成的结果及时保存到tmp文件，事物提交时，将tmp文件改成正式文件。灾难发生时，对已经处理过的数据，使用已经保存的tmp文件里的结果，不再重新处理。查重有两种方法，暂时使用第1种，对第2种的支持以后再加进去：1、 查重现在本身有事物控制，不需要在查重外另做控制。灾难时，删除tmp

9、文件，启动进程，就可以了。2、 查重提交一次索引文件作为一个事物，提交完索引文件时，发出一个事物完成消息，由查重之外的模块完成事物控制（删除临时文件）。一次事物可能是处理多个文件。批价和出帐一条话单作为一个事物，事物完成不需发要事物完成消息。批价和出帐后面会挂一个输出模块，把话单及时保存到tmp文件。其它模块各模块的处理在billing40/framefile/frame_file_interface.h里，有两个类, CProcessFileTherad和CProcessCdrThread，这两类都是从CThread派生。处理文件的模块都从CProcessFileThread派生，处理话单的

10、模块都从CProcessCdrThread派生。解码要做的事情1、 用输入的pFrameFile参数调用get_inputFileDesc（）获得一个输入文件的句柄，使用这个句柄读取输入文件的内容。2、 用输入的pFrameFile参数调用const char *get_inputFileName() const获得输入文件名。3、 解码，每解一条话单，new一个CXdr，解码出来的xdr，设置一些属性，并把它放入文件。参考下面代码。4、 最后，用send_data把话单送到下一模块。#include class CDecodeThread : public bs_frame_file:CPr

11、ocessFileThreadpublic:int32 init();int32 destroy();int32 run(bs_frame_file:CFrameFile* pFrameFile)Int32 iFileDesc = pFrameFile -get_inputFileDesc();for()Bs_frame_file:CXdr *pXdr = pFrameFile - get_newXdr();If(decode(xdr, buf) != 0)/解码失败pFrameFile -delete_xdr(pXdr); continue;/解码成功if(normal)/正确话单send_

12、data(pXdr);else/错单send_data(pXdr, 1);pFrameFile - end_decodeFile ();return 0;private:;bs_frame:CThread *create_instanceDecode()Return new CDecodeThread ();查重要做的事情1、从输入的pFrameFile, 通过get_xdrBegin()，get_xdrEnd()和get_xdrNext（）获得每条话单。#include class CDupThread : public bs_frame_file: CProcessFileThreadpu

13、blic:int32 init();int32 destroy();int32 run(bs_frame_file:CFrameFile* pFrameFile)Bs_frame_file:CXdr *pTmpXdr=NULL; Bs_frame_file:CXdr *pEndXdr= pFrameFile -get_xdrEnd();For(Bs_frame_file:CXdr *pXdr= pFrameFile -get_xdrBegin();pXdr!= pEndXdrpXdr= pTmpXdr)Check_dup();pTmpXdr= pFrameFile -get_xdrNext(p

14、Xdr)；If(pTmpXdr = pEndXdr)Commit index file; /提交查重索引文件if(normal)/正确话单send_data(pXdr);else/错单send_data(pXdr, 1);return 0;private:;bs_frame:CThread *create_instanceDup()Return new CDupThread ();批价要做的事情(还有业务分析)1、run的输入是一个xdr，直接开始批价。2、处理时，如果需要拆话单，先从xdr里获得文件的句柄，再调用它的add_record函数。如下面代码。#include class CRat

15、ingThread : public bs_frame_file:CProcessCdrThreadpublic:int32 init();int32 destroy();int32 run(Bs_frame_file:CXdr& xdr)rating.();if(add)/要拆话单bs_frame_file:CFrameFile *pFrameFile = (CFrameFile *)xdr.get_attrInfo()-get_info();Bs_frame_file:CXdr *pXdr = pFrameFile - get_newXdr(&xdr);/process pXdrpFram

16、eFile-add_record(pXdr); if(normal)/正确话单send_data(pXdr);else/错单send_data(pXdr, 1);if(normal)/正确话单send_data(&xdr);else/错单send_data(&xdr, 1);return 0;private:;bs_frame:CThread *create_instanceRating()Return new CRatingThread ();入库要做的事情入库的输入在上一版本的文档里是一个xdr。但是入库需要按文件提交事物，输入的xdr也不能按照话单在文件里的顺序输入，难以确定最后一条话单

17、。改成以文件为单位输入。1、 从输入的pFrameFile, 通过get_xdrBegin()，get_xdrEnd()和get_xdrNext（）获得每条话单。如下面代码。#include class CDataLoaderThread : public bs_frame_file: CProcessFileThreadpublic:int32 init()M_iIndexLaterLink = CXdrList:get_fieldIndex (“LATER_LINK”);int32 destroy();int32 run(bs_frame_file:CFrameFile* pFrameFi

18、le)For(Bs_frame_file:CXdr *pXdr= pFrameFile -get_xdrBegin();pXdr!= pFrameFile -get_xdrEnd();pXdr= pFrameFile -get_xdrNext(pXdr)Insert_table(pXdr);If(newxdr) Bs_frame_file:CXdr *pNewXdr = pFrameFile - get_newXdr(*pXdr); Insert_table(pNewXdr);Send_data(pNewXdr);If(last record)Commit; /提交事物if(normal)/正

19、确话单send_data(pXdr);else/错单send_data(pXdr, 1);return 0;private:;bs_frame:CThread *create_instanceDataLoader()Return new CDataLoaderThread ();数据管理中心要做的事情在int32 run(bs_frame:CData *p)里完成日常维护的事情。函数里要有一个sleep，不然会导致线程CPU占用过高。p指向NULL，对数据管理中心没用。在int32 control(DOMDocument& resultDoc, const DOMDocument& contr

20、olDoc)里处理前台发过来的控制消息。#include class Codmc : public bs_frame:CThreadpublic:int32 init(const char *cszCfg);int32 destroy();int32 run(bs_frame:CData *p)/日常维护Sleep(1);return 0;int32 control(DOMDocument& resultDoc, const DOMDocument& controlDoc);private:;bs_frame:CThread *create_instanceOdmc()Return new C

21、odmc ();每个业务处理模块要做的事情int32 init();程序初始化时调用。如读配置项，连接MDB，获取xdr的index，装载局数据等int32 destroy();程序退出前在这个函数做一些收尾工作。如，从MDB断开。int32 control(DOMNode & resultDoc, const DOMNode & controlDoc);前台发过来控制消息。DOMDocument是xml的文档类。controlDoc是控制消息，里面有控制命令，控制参数，控制时要使用的一些数据。resultDoc是控制结果。Bs_frame_file:CXdr, 从CXdr派生，在CXdr的基

22、础上增加了框架管理使用的一些属性，业务处理模块不需要关心。像使用CXdr一样使用bs_frame_file:CXdr就可以了。数据管理中心使用CXdr不需要使用bs_frame_file:CXdr。文件的处理使用了文件模块，头文件为“#include ”，动态库名字为libFrameFile$(DLLTAIL)，编译时用参数-lFrameFile$(BUILDTYPE)。数据管理中心不使用这个lib。注意事项1. 因为一个节点可能会处理多种话单，但使用xdr的时候是字段的名字只对应一个index。需要xdr支持这种对应关系。2. 有一些函数不是多线程安全的，不能在多线程里使用，如：a) Cti

23、me，localtime，gmtime，asctime，要改成使用ctime_r，localtime_r，gmtime_r，asctime_r3. init函数最好写成可以被多次调用的方式，如改了配置文件，前台通知节点更新配置项的值，但不需要停止进程。4. 写日志方式，使用起来也不方便，请大家提提建议。5. frame_interface.h里定义了一个枚举类型STAT，这个名字和系统下一个宏同名（在/usr/include/sys/dir.h），改成THREAD_STAT。6. 编译时可能会碰到类似下面的错误，是因为ailog.h里定义了宏DEBUG_LEVEL和frame_interfac

24、e.h里冲突之故，新框架下不再使用ailog，将ailog.h的包含去掉后，就可以编译通过了。/data03/home/bill401/include/billing40/frame_interface.h, line 53.9: 1540-0063 (S) The text 0 is unexpected.7. 如果程序运行出现莫明其妙的错误，可能是bill402上的程序和lib不够新，到bill401上把所有的程序和lib拷过来，覆盖bill402上的。再试。8. 新的文件格式里有文件头信息，如果拿老框架跑出来的文件处理的话，要对这些文件做一些处理（加上文件头信息），新框架下提供了一个程序

25、，change_file，如change_file f DASE1070418.TTFILE00.8887.roam t dr_gsm，把老话单文件转成新话单文件。解码的输入文件不需要做转化。9. LogAppend函数，框架提供两个LogAppend函数，一个是类成员class CThread void LogAppend(const LOG_LEVEL& iLevel, const char *cszFormat, .) const; ，还有一个是全局函数void LogAppend(const char *cszModule, const LOG_LEVEL& eLogLevel, co

26、nst char *cszFormat, .);，使用成员函数打日志时，框架会标识是哪个线程打的日志，用全局函数打日志时，框架只用用函数的参数（cszModule）标识是谁打的日志。建议使用成员的函数打日志，由框架增加日志来源的标识，在一个进程里只有一个实例的模块，可以用全局函数打日志，用(cszModule)框架日志来源。新的LogAppend函数的用法和ailog里的LogAppend的用法有改变，ailog的做法是，先声明一个BUF，再格式化数据，然后调用LogAppend，新的做法是，格式化数据放在LogAppend里做，去掉了ailog里使用的messageFile.txt，增加了名

27、字空间NAMESPACE_BILLING40_FRAME，使用全局的LogAppend和日志级别时，都要加有名字空间。Ailog的LogAppend一般是这样打日志的：Char szBuf1024;Sprintf(szBuf, E302:file:%s,has zero record, taskStat.get_fileName().data();LogAppend(WARN_LEVEL, “DebugInfo”, szBuf);新框架下是这样打日志的：成员方式：This-LogAppend(NAMESPACE_BILLING40_FRAME WARN_LEVEL, E302:file:%s,

28、has zero record, taskStat.get_fileName().data();全局函数方式： NAMESPACE_BILLING40_FRAME LogAppend(“rating”, NAMESPACE_BILLING40_FRAME WARN_LEVEL, E302:file:%s,has zero record, taskStat.get_fileName().data();10.例子时钟文件清单Time.h#ifndef _BILLING_TIME_H_#define _BILLING_TIME_H_#include frame_interface.hclass CT

29、imeMgr : public CThreadpublic: int32 init(const char *cszCfg); int32 destroy(); int32 run(CData *p); CTimeMgr(); virtual CTimeMgr();private: enum TIME_STEP = 10, ARR_SIZE = 3600*TIME_STEP ; CTime m_arrTimeARR_SIZE; int32 m_iIndex;#ifdef _cplusplusextern C #endifCThread *create_clockInstance();#ifdef

30、 _cplusplus#endif#endif /_BILLING_TIME_H_Time.cpp#include time.hCTimeMgr:CTimeMgr() : m_iIndex(0)CTimeMgr:CTimeMgr()int32 CTimeMgr:init(const char *cszCfg) m_iIndex = 0; run(NULL); return 0;int32 CTimeMgr:destroy() return 0;int32 CTimeMgr:run(CData *p) CTime *pTime = &m_arrTimem_iIndex; time(&pTime-

31、m_lTime); memset(&pTime-m_tm, 0, sizeof(&pTime-m_tm); localtime_r(&pTime-m_lTime, &pTime-m_tm); sprintf(pTime-m_szYYYYMMDDH24MISS, %.4d%.2d%.2d%.2d%.2d%.2d, pTime-m_tm.tm_year + 1900, pTime-m_tm.tm_mon + 1, pTime-m_tm.tm_mday, pTime-m_tm.tm_hour, pTime-m_tm.tm_min, pTime-m_tm.tm_sec); memset(pTime-m

32、_szYYYYMMDD, 0, sizeof(pTime-m_szYYYYMMDD); strncpy(pTime-m_szYYYYMMDD, pTime-m_szYYYYMMDDH24MISS, 8); g_pTime = pTime; if(+m_iIndex = ARR_SIZE) m_iIndex = 0; fd_set readSet; FD_ZERO(&readSet); FD_SET(0, &readSet); struct timeval tm; tm.tv_sec = 0; tm.tv_usec = 1000/TIME_STEP; select(1, &readSet, NU

33、LL, NULL, &tm); return 0;#ifdef _cplusplusextern C #endifCThread *create_clockInstance() return new CTimeMgr();#ifdef _cplusplus#endifXdr写文件xdr_write_file.h#ifndef _XDR_WRITE_FILE_H_#define _XDR_WRITE_FILE_H_#include bill_file.h#include ./record.hclass CXdrWriteFile : public CProcessCdrThreadpublic:

34、 enum RECORD_BUF_LEN = 1024 *1024 ;public: virtual int32 init(const char *cszCfg); virtual int32 destroy(); CXdrWriteFile(); virtual CXdrWriteFile();protected: virtual int32 run(CXdr& xdr); virtual int32 exception(CXdr& xdr);private:;#ifdef _cplusplusextern C #endifCThread *create_writeFileInstance(

35、);#ifdef _cplusplus#endif#endif /_XDR_WRITE_FILE_H_文件清单xdr_write_file.cpp#include xdr_write_file.h#include #include #include CXdrWriteFile:CXdrWriteFile()CXdrWriteFile:CXdrWriteFile()int32 CXdrWriteFile:init(const char *cszCfg) int32 iRet = 0; return CProcessCdrThread:init(cszCfg);int32 CXdrWriteFil

36、e:destroy() int32 iRet = 0; return iRet;int32 CXdrWriteFile:run(CXdr& xdr) xdr.m_bFinished = true;/ AISTD cout -write get date: pFrameFile AISTD endl get_info(); if(pFrameFile=NULL) LogAppend(FATAL_LEVEL, the bill file in xdr is null); return 0; int32 iRet = pFrameFile-write_record(xdr, m_cszNodeNam

37、e); if(iRet = 1) /file finished send_data(pFrameFile); return 0;int32 CXdrWriteFile:exception(CXdr& xdr) xdr.m_bFinished = true; return 0;#ifdef _cplusplusextern C #endifCThread *create_writeFileInstance() return new CXdrWriteFile();#ifdef _cplusplus#endif业务分析配置举例配置文件采用xml格式文件后，是这样的。10.10.10.1722550

38、0rule/analyse_gsm_zj1.exprrule/analyse_gsm_zj2.expr$(OB_REL)/lib/libExprAnalyseD1.soRegAnalyseCommExprFun=$(OB_REL)/lib/libExprAnalyseD2.soRegAnalyseCommExprFun2sTHIS_PROV=571iDEFAULT_PRODUCT_ID=50001001iCALLER_OTHER_CHECK=1iCALLER_OTHER_ZERO_BEFOR=1iCALLEE_OTHER_CHECK=0iSEEK_USER_INFO=1iCHECK_NO_OW

39、NER=0iCHECK_USER_INFO_SUB_TIME=1iOPP_INFO_FLAG=0sRATE_PROD_JPAN_KOREA=90134001iPLAN_ID_JPAN_KOREA=9000sRATE_PROD_PP_USER=50001018iPLAN_ID_PP_USER=600iTOLL_SHORT_DURATION=3s2i_RIN_SHORTEST_EXCEPT_HEAD=17950;17910;17930;17908;17920;17900;17990iBOSS_VPMN_GROUPTYPE=5iVPMN_CAST_FLAG=1iDO_FILTER_ERR=0iCHE

40、CK_USER_INFO_SERVICE_ID=50001sEXPIRE_TIME_1860=20061218000000s2s_SIMN_I_FILE_PLMNPROMLIST=HKGPP:50103500;程序里写成这样class CConfigpublic: class CExprFileConfig public: char m_szExprFileName512; READ_CONFIG_BEGIN READ_VALUE(m_szExprFileName, file, , NAMESPACE_BILLING40_FRAME FATAL_LEVEL) READ_CONFIG_END ;

41、 typedef NAMESPACE_BILLING40_FRAME CConfigList ExprFileList; class CLibFuncConfig public: char m_szLibName512; char m_szFuncName128; READ_CONFIG_BEGIN READ_VALUE(m_szLibName, LOAD_OPER_LIB, , NAMESPACE_BILLING40_FRAME FATAL_LEVEL) READ_VALUE(m_szFuncName, LOAD_OPER_FUNC, , NAMESPACE_BILLING40_FRAME

42、FATAL_LEVEL) READ_CONFIG_END ; typedef NAMESPACE_BILLING40_FRAME CConfigList LibFuncList; class CGlobalVarConfig public: char m_szValue1024; READ_CONFIG_BEGIN READ_VALUE(m_szValue, value, , NAMESPACE_BILLING40_FRAME FATAL_LEVEL) READ_CONFIG_END ; typedef NAMESPACE_BILLING40_FRAME CConfigList GlobalV

43、arList; char m_szCustomerServerHost16; uint16 m_usCustomerServerPort; ExprFileList m_lstExprFile; LibFuncList m_lstLibFunc; GlobalVarList m_listGlobalVar;READ_CONFIG_BEGIN READ_VALUE(m_szCustomerServerHost, customer_server_host, , NAMESPACE_BILLING40_FRAME FATAL_LEVEL) READ_VALUE(m_usCustomerServerP

44、ort, customer_server_port, , NAMESPACE_BILLING40_FRAME FATAL_LEVEL) READ_OBJ(m_lstExprFile, expr_file, NAMESPACE_BILLING40_FRAME FATAL_LEVEL) READ_OBJ(m_lstLibFunc, reg_oper_lib_set, NAMESPACE_BILLING40_FRAME FATAL_LEVEL) READ_OBJ(m_listGlobalVar, reg_global_var, NAMESPACE_BILLING40_FRAME FATAL_LEVE

45、L)READ_CONFIG_END;读取配置文项class CAnalyseThread : public bs_frame_file:CProcessCdrThreadpublic:int32 init()Int32 iRet = read_config(m_objConfig);If(iRet = 0) Return 0;ElseReturn -1;int32 destroy();int32 run(Bs_frame_file:CXdr& xdr)private:CConfig m_objConfig;配置文件举例红色部分需要按照模块情况修改成具体的配置。解码的配置文件1libFrameFileD.socreate_inputFileInstence2libFrameFileD.socreate_outputRecordInstence3libFrameFileD.socreate_outputFileInstence4decode lib namedecode func name1input112output record213output file314decode