基于新框架的開發(fā)多線程框架下開發(fā)

1、框架下開發(fā)基于新框架的開發(fā)目錄結(jié)構(gòu)billing40/frame/svr/xdrframe/frame_interface.h 框架billing40/app/framefile/frame_file_interface.h文件(輸入，輸入，事物控制，狀態(tài)) billing40在products/openboss/billing40用:cvs_co products/openboss/billing40cd products/openboss/billing40cvs_up r ob20_dev開發(fā)如批價(jià)：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 所有在框架下運(yùn)行的lib都有一個(gè)class，這個(gè)class直接或間接繼承cthreadl 在這個(gè)class里，須實(shí)現(xiàn)虛擬函數(shù)

3、run。如：對(duì)批價(jià)來說run的參數(shù)p就是一個(gè)xdr的指針l lib里還須提供一個(gè)函數(shù)，創(chuàng)建這個(gè)class的對(duì)象，如上面create_instance，框架為每個(gè)線程創(chuàng)建一個(gè)class的對(duì)象。l 每個(gè)線程一個(gè)cratingthread對(duì)象，在cratingthread里的數(shù)據(jù)都是線程安全的l 處理完成的數(shù)據(jù)，調(diào)用send_data(void *p, const int32& iindex)，把數(shù)據(jù)送到下一個(gè)節(jié)點(diǎn)（模塊）。參數(shù)iindex，是在當(dāng)前節(jié)點(diǎn)的處理結(jié)果要分開輸出到不同節(jié)點(diǎn)時(shí)用。如正確話單iindex=0, 錯(cuò)單iindex=1。l init函數(shù)。線程初始化。如讀配置文件，連接到mdbse

4、rver等。l 時(shí)鐘，ctime *g_ptime，每100微秒更新一次。ctime里有多格式的時(shí)間，可以直接使用，如批價(jià)使用的process_time。ctime的聲明見下面l 配置文件采用xml，而且原來有一些配置到數(shù)據(jù)庫表里的配置也移到了配置文件，框架和數(shù)據(jù)庫沒有關(guān)系了，不再提供otl_connect，也沒有了數(shù)據(jù)庫帳號(hào)，讀取配置項(xiàng)統(tǒng)一用cthread里的read_config ?？蚣芤呀?jīng)對(duì)配置文件進(jìn)行了初始化，各模塊不需要再對(duì)配置文件做初始化，直接讀取即可?？蚣芴峁┝艘恍┳x取配置項(xiàng)的宏，可以方便地讀取到配置項(xiàng)。如業(yè)務(wù)分析的配置，見下面舉例。l 多線程框架在名字空間bs_frame里l

5、包含頭文件frame_interface.h，編譯時(shí)使用-lframedl exception函數(shù)是異常時(shí)的處理，如批價(jià)，在異常時(shí)，把話單打成錯(cuò)單。l 寫日志統(tǒng)一采用cthread:logappend，框架可以會(huì)把一些日志發(fā)到前臺(tái)。同時(shí)，通過這個(gè)函數(shù)寫日志時(shí)，框架會(huì)自動(dòng)的寫日志的線程所在的流水線和節(jié)點(diǎn)的信息加上。這也帶來了一個(gè)麻煩，寫日志時(shí)必需有一個(gè)cthread對(duì)象(業(yè)務(wù)處理模塊都有一個(gè)類（a）從cthread派生，在這個(gè)類里可以直接調(diào)用logappend，在其它不是從cthread類派生的類里，如果需要寫日志，需要把a(bǔ)的指針傳給它，以使這些類也可以寫日志。)l 節(jié)點(diǎn)級(jí)變量，當(dāng)節(jié)點(diǎn)內(nèi)的所有線

6、程使用相同的變量時(shí)，變量可以放到節(jié)點(diǎn)里，變量在節(jié)點(diǎn)內(nèi)只有一份，節(jié)點(diǎn)級(jí)變量只能被設(shè)置一次。class ctimepublic: struct timeval m_tmval; / m_tmval. tv_sec從1970年到當(dāng)前時(shí)間的秒數(shù), 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位數(shù)的微秒 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);事物控制事物控制總的指導(dǎo)思想是，對(duì)有需要做事物控制的模塊，將其處理完成的結(jié)果及時(shí)保存到tmp文件，事物提交時(shí)，將tmp文件改成正式文件。災(zāi)難發(fā)生時(shí)，對(duì)已經(jīng)處理過的數(shù)據(jù)，使用已經(jīng)保存的tmp文件里的結(jié)果，不再重新處理。查重有兩種方法，暫時(shí)使用第1種，對(duì)第2種的支持以后再加進(jìn)去：1、 查重現(xiàn)在本身有事物控制，不需要在查重外另做控制。災(zāi)難時(shí)，刪除tmp

9、文件，啟動(dòng)進(jìn)程，就可以了。2、 查重提交一次索引文件作為一個(gè)事物，提交完索引文件時(shí)，發(fā)出一個(gè)事物完成消息，由查重之外的模塊完成事物控制（刪除臨時(shí)文件）。一次事物可能是處理多個(gè)文件。批價(jià)和出帳一條話單作為一個(gè)事物，事物完成不需發(fā)要事物完成消息。批價(jià)和出帳后面會(huì)掛一個(gè)輸出模塊，把話單及時(shí)保存到tmp文件。其它模塊各模塊的處理在billing40/framefile/frame_file_interface.h里，有兩個(gè)類, cprocessfiletherad和cprocesscdrthread，這兩類都是從cthread派生。處理文件的模塊都從cprocessfilethread派生，處理話單的

10、模塊都從cprocesscdrthread派生。解碼要做的事情1、 用輸入的pframefile參數(shù)調(diào)用get_inputfiledesc（）獲得一個(gè)輸入文件的句柄，使用這個(gè)句柄讀取輸入文件的內(nèi)容。2、 用輸入的pframefile參數(shù)調(diào)用const char *get_inputfilename() const獲得輸入文件名。3、 解碼，每解一條話單，new一個(gè)cxdr，解碼出來的xdr，設(shè)置一些屬性，并把它放入文件。參考下面代碼。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/錯(cuò)單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/錯(cuò)單send_data(pxdr, 1);return 0;private:;bs_frame:cthread *create_instancedup()return new cdupthread ();批價(jià)要做的事情(還有業(yè)務(wù)分析)1、run的輸入是一個(gè)xdr，直接開始批價(jià)。2、處理時(shí)，如果需要拆話單，先從xdr里獲得文件的句柄，再調(diào)用它的add_record函數(shù)。如下面代碼。#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/錯(cuò)單send_data(pxdr, 1);if(normal)/正確話單send_data(&xdr);else/錯(cuò)單send_data(&xdr, 1);return 0;private:;bs_frame:cthread *create_instancerating()return new cratingthread ();入庫要做的事情入庫的輸入在上一版本的文檔里是一個(gè)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 (“l(fā)ater_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/錯(cuò)單send_data(pxdr, 1);return 0;private:;bs_frame:cthread *create_instancedataloader()return new cdataloaderthread ();數(shù)據(jù)管理中心要做的事情在int32 run(bs_frame:cdata *p)里完成日常維護(hù)的事情。函數(shù)里要有一個(gè)sleep，不然會(huì)導(dǎo)致線程cpu占用過高。p指向null，對(duì)數(shù)據(jù)管理中心沒用。在int32 control(domdocument& resultdoc, const domdocument& contr

20、oldoc)里處理前臺(tái)發(fā)過來的控制消息。#include class codmc : public bs_frame:cthreadpublic:int32 init(const char *cszcfg);int32 destroy();int32 run(bs_frame:cdata *p)/日常維護(hù)sleep(1);return 0;int32 control(domdocument& resultdoc, const domdocument& controldoc);private:;bs_frame:cthread *create_instanceodmc()return new c

21、odmc ();每個(gè)業(yè)務(wù)處理模塊要做的事情int32 init();程序初始化時(shí)調(diào)用。如讀配置項(xiàng)，連接mdb，獲取xdr的index，裝載局?jǐn)?shù)據(jù)等int32 destroy();程序退出前在這個(gè)函數(shù)做一些收尾工作。如，從mdb斷開。int32 control(domnode & resultdoc, const domnode & controldoc);前臺(tái)發(fā)過來控制消息。domdocument是xml的文檔類。controldoc是控制消息，里面有控制命令，控制參數(shù)，控制時(shí)要使用的一些數(shù)據(jù)。resultdoc是控制結(jié)果。bs_frame_file:cxdr, 從cxdr派生，在cxdr的基

22、礎(chǔ)上增加了框架管理使用的一些屬性，業(yè)務(wù)處理模塊不需要關(guān)心。像使用cxdr一樣使用bs_frame_file:cxdr就可以了。數(shù)據(jù)管理中心使用cxdr不需要使用bs_frame_file:cxdr。文件的處理使用了文件模塊，頭文件為“#include ”，動(dòng)態(tài)庫名字為libframefile$(dlltail)，編譯時(shí)用參數(shù)-lframefile$(buildtype)。數(shù)據(jù)管理中心不使用這個(gè)lib。注意事項(xiàng)1. 因?yàn)橐粋€(gè)節(jié)點(diǎn)可能會(huì)處理多種話單，但使用xdr的時(shí)候是字段的名字只對(duì)應(yīng)一個(gè)index。需要xdr支持這種對(duì)應(yīng)關(guān)系。2. 有一些函數(shù)不是多線程安全的，不能在多線程里使用，如：a) cti

23、me，localtime，gmtime，asctime，要改成使用ctime_r，localtime_r，gmtime_r，asctime_r3. init函數(shù)最好寫成可以被多次調(diào)用的方式，如改了配置文件，前臺(tái)通知節(jié)點(diǎn)更新配置項(xiàng)的值，但不需要停止進(jìn)程。4. 寫日志方式，使用起來也不方便，請(qǐng)大家提提建議。5. frame_interface.h里定義了一個(gè)枚舉類型stat，這個(gè)名字和系統(tǒng)下一個(gè)宏同名（在/usr/include/sys/dir.h），改成thread_stat。6. 編譯時(shí)可能會(huì)碰到類似下面的錯(cuò)誤，是因?yàn)閍ilog.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. 如果程序運(yùn)行出現(xiàn)莫明其妙的錯(cuò)誤，可能是bill402上的程序和lib不夠新，到bill401上把所有的程序和lib拷過來，覆蓋bill402上的。再試。8. 新的文件格式里有文件頭信息，如果拿老框架跑出來的文件處理的話，要對(duì)這些文件做一些處理（加上文件頭信息），新框架下提供了一個(gè)程序

25、，change_file，如change_file f dase1070418.ttfile00.8887.roam t dr_gsm，把老話單文件轉(zhuǎn)成新話單文件。解碼的輸入文件不需要做轉(zhuǎn)化。9. logappend函數(shù)，框架提供兩個(gè)logappend函數(shù)，一個(gè)是類成員class cthread void logappend(const log_level& ilevel, const char *cszformat, .) const; ，還有一個(gè)是全局函數(shù)void logappend(const char *cszmodule, const log_level& eloglevel, co

26、nst char *cszformat, .);，使用成員函數(shù)打日志時(shí)，框架會(huì)標(biāo)識(shí)是哪個(gè)線程打的日志，用全局函數(shù)打日志時(shí)，框架只用用函數(shù)的參數(shù)（cszmodule）標(biāo)識(shí)是誰打的日志。建議使用成員的函數(shù)打日志，由框架增加日志來源的標(biāo)識(shí)，在一個(gè)進(jìn)程里只有一個(gè)實(shí)例的模塊，可以用全局函數(shù)打日志，用(cszmodule)框架日志來源。新的logappend函數(shù)的用法和ailog里的logappend的用法有改變，ailog的做法是，先聲明一個(gè)buf，再格式化數(shù)據(jù)，然后調(diào)用logappend，新的做法是，格式化數(shù)據(jù)放在logappend里做，去掉了ailog里使用的messagefile.txt，增加了名

27、字空間namespace_billing40_frame，使用全局的logappend和日志級(jí)別時(shí)，都要加有名字空間。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();全局函數(shù)方式： namespace_billing40_frame logappend(“rating”, namespace_billing40_frame warn_level, e302:file:%s,has zero record, taskstat.get_filename().data();10.例子時(shí)鐘文件清單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業(yè)務(wù)分析配置舉例配置文件采用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_