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

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 fatal_level) read_config_