{业务管理}业务数据集成

1、业务管理）业务数据集成功能概述业务数据集成是指 MDMServer 和外围系统之间的数据交换时需要中间处理的模块应 用层。如图所示，外围系统主要通过三种模式和 MDMServer 交互 （这里不考虑 ETL）：直接的 webservice 访问主要通过 http 协议联机访问，壹般不通过业务集成层而 直接访问数据服务层近实时的消息通知主要通过 MQ 消息访问，需要于业务集成层进行必要的处理 批量处理主要通过文件进行数据传递，需要于业务集成层进行处理 业务集成服务层的主要内容包括：近实时的消息通知数据标准代码的转换交易 xml 的拆分其他规则批量处理标准代码的转换入库方式判断处理，判断是否是新增

2、仍是更改交易拆分其他规则1.1.1. 联机数据集成（基于 MQ/XML ）联机数据集成主要处理近实时消息通知，处理模式如下图所示。MQ 组件是所有的客户信息更新消息的载体， MsgReceiver 组件负责接收 MQ 的消息， 通过 MsgParser 组件进行 xml 格式解析， MsgReceiver 转交到 IntegrateFW 组件，完成业IntegrateFW 组件是转发框架，负责业务集成层的基本控制，根据报文的类型调用必要的规 则处理进行数据转换或者信息拆分，然后交由具体逻辑处理实现模块 IntegrateImpl 务集成逻辑，且最终通过 ServiceInvocation 模块

3、调用 MDM 的服务，完成交易。数据服务层主要包括扩展服务和组合服务。如上图，根据业务需求，壹般性的顺序如下详述：1ejbCreate MQ 中的信息通知 WAS 容器，根据 MsgReceiver 组件的 MDB 情况，创 建壹个实例进入池态；2onMessage WAS 容器调用 MDB 的 onMessage 方法，且把消息作为参数传递到该 方法；2.1deliver onMessage 方法中调用 IntegrateFW 组件的 deliver 方法，且传递消息；IntegrateFW 组件调用 MsgParser 组件解析传入的消息体；MsgParser 组件根据消息体解析，生成壹个

4、 IBObj 对象；parse 方法返回生成的 IBObj 对象；获取IBObj 对象的服务类型；返回字符串结果；获取 IBObj 的请求者；返回；IntegrateFW 根据服务对象和请求者，实例化壹个 IntegrateImpl 组件中集成服务实例；调用具体实例的业务逻辑，且把 IBObj 作为参数传递；调用标准代码转换规则，进行必要规则处理；返回规则处理后的 IBObj 对象；调用客户识别规则，进行识别调用返回于具体的业务逻辑处理中， 根据具体需要调用 MsgParser 创建 需要的 IBObj 对象；MsgParser 构建新的 IBObj 实例；返回具体的 IBObj 实例；处理具

5、体的 IBObj 实例内容；处理内容返回；处理完所有的 IBObj 实例内容后， 调用方法转换为 xml 标准服务 格式；返回 xml 字节流；调用 MDM 服务；返回；完成具体业务集成逻辑实例的执行；2.2deliver 完成整个业务集成逻辑的执行；注：到 之间可能有多次的 callRule 和 invoke 调用；到 的调用需要包括于事务处理中；批量处理的业务集成层设计参见批量型章节。. MsgReceiver 组件MsgReceiver 组 件 主 要 由 IntegrateMDB 组 成 ， IntegrateMDB 是 壹 个 MessageDriverBean ，通过

6、 onMessage(Message) 方法接收 MQ 的消息通知。信息接收组件主要负责从 MQ 接收消息， 参数 Message 使用 BytesMessage 类型， 消 息格式为 XML 形式， 其具体的接口定义参见 服务接口定义 中的关联通过 MQ 接口部分。接口主要包括：个险增加客户 (addPerson/CSCMQ)个险保全修改客户 (updatePerson/CSCMQ)电商网上销售增加客户 (addPerson/TOLMQ)电商网上销售修改客户基本信息 (updatePersonBase/TOLMQ)电商网上销售保全修改客户 (updatePerson/TOLMQ)注：之上括号

7、中的服务名是指各个业务系统提交到 MQ 中的消息服务名，而不是 MDM 调用的标准服务名；组件调用 IntegrateFW 组件进行后续处理。publicvoidonMessage(Messagemsg)if(msginstanceofBytesMessage)BytesMessagebm=(BytesMessage)msg;longlength=bm.getBodyLength();bytebs=newbytelength;bm.getBytes(bs);BytesInputStreambis=newBytesInputStream(bs);IntegrateFW.deliver(bis);

8、else. IntegrateFW 组件IntegrateFW 组件是框架组件， 负责业务集成层的控制调度。 组件负责调用 MsgParser 组件，解析 xml 报文，然后根据报文类型调用具体的控制处理逻辑，同时调用业务处理规则 进行数据和 业务 处理，最后形 成处理后的报文提交 ServiceInvocation 组件调用 MDMService ，完成整个处理，同时如果于处理中发生异常，则交由 IntegException 组件 处理异常结果。IntegrateFW 组件中对于多个 service 调用作为同壹事务来管理，如果调用出错需要进 行回滚处理。IntegrateFW

9、 组件暴露壹个接口 IntegrateLogic 供 IntegrateImpl 组件实现，其接口 方法如下。* 执行具体的业务集成任务 .用于 MQ 异步通知模式* 于 IntegrateImpl 组件实现该方法，处理具体某壹项接口的逻辑调用，壹般包括* 代码转换、服务拆分、服务变换、 DSP 判断等*paramis 输入流，是通过 MQ 接收到的消息内容， xml 字节流*throwsItegrateException 例外，如果发生例外需要到例外组件处理*/publicvoidexecute(IBobjobj)throwsItegrateException;组件对外调用的 Fa?ade

10、接口调用是 IntegratFW 类，使用静态方法。 /* 传递到 IntegrateFW 组件执行后续任务 .用于 MQ 异步通知模式*paramis 输入流，是通过 MQ 接收到的消息内容， xml 字节流*/publicstaticvoiddeliver(InputStreamis);. IntegrateImpl 组件IntegrateImpl 组件是根据报文类型确定的具体的业务集成逻辑的实现。 壹期需求主要包括：个险增加客户个险修改客户基本信息个险保全修改客户信息网上销售增加客户参见个险增加客户，只是其接口内容稍有差异 网上销售修改客户基本信息 参见个险修改客户基本信息

11、，只是其接口内容稍有差异 网上销售保全修改客户信息 参见个险保全修改客户信息，只是其接口内容稍有差异此组件需要处理地址、电话、电子邮件、客户标识等的格式变换，是把输入的对象格式 变为标准的 MDM 扩展服务格式，包括：从粗粒度服务格式转换为 updatePersonName 服务格式转 换 为 changePartyAddress 服 务 格 式 服 务 中 确 定 具 体 调 用 correctPartyAddress 仍是 addPartyAddress 仍是不做处理；转 换 为 changePartyContactMethod 服 务 格 式 服 务 中 定 具 体 调 用updateP

12、artyContactMethod 仍是 addPartyContactMethod 仍是不做处理； 从粗粒度服务格式转换为 changePartyIdentifer 服务格式服务中定具体调用 updatePartyIdentifer 仍是 addPartyIdentifer 仍是不做处理； 具体接口参见服务接口和 MDM 开发文档。 服务逻辑参见关联于线服务组合服务设计；. MsgParser 组件信息解析主要是根据 xml 报文解析其内容。MsgParser 组件根据不同的报文类型解析不同的内容。MsgParser 组件仍负责输出变更后的报文结果。MsgParser 是暴露于

13、外的 Fa?ade 调用界面，通过俩个个方法接收外部调用。 /* 解析 xml.*paramis 输入流，是通过 MQ 接收到的消息内容， xml 字节流*/publicstaticIBObjparse(InputStreamis);/* /* * 设置 IBObj 的类型 .是指对象类型，如 TCRMPersonBObj 、*TCRMAdminContEquivBObj、TCRMPartyAddressBObj 、TCRMAddressBObj*/publicvoidsetType(Stringname);publicStringgetType();/*根据名称构建新的 IBObj 对象 .

14、*paramname 输入的 IBObj 名称*return 返回 IBObj 对象*/publicstaticIBObjconstructIBObj(Stringname);IBObj 是用于定义报文的 bean ，类似 MDM 中的 BObj 。等。IBObj 主要定义 xml 中的 bobj 对象， IBObj 内部能够嵌套。其暴露的接口如下。publicvoidsetTCRMTxObject(Stringobj);publicStringgetTCRMTxObject();/* 设置 IBObj 对象 .*/publicvoidsetIBObj(IBObjobj);/*根据名字获取 I

15、BObj 对象 .只能获取下壹级对象*/publicIBObjgetIBObj(Stringtype);publicStringgetAttribute(Stringtype);/* 设置属性 .自动区分处理 Extension 属性 .*/publicStringsetAttribute(Stringtype,Stringvalue);/*设置服务头*/publicvoidsetHeader(Stringheader);publicStringgetHeader(); 设置请求控制部分*/publicvoidsetRequestControl(Stringrc);publicStringge

16、tRequestControl();/*设置请求控制部分的请求名称 .*/publicvoidsetRequestName(Stringrn);publicStringgetRequestName();/*设置请求控制部分的 LOB.*/publicvoidsetRequestLOB(Stringrlob);publicStringgetRequestLOB();. DSPRule 组件处理 DSP 规则，具体参见 DSP 设计中于线可疑客户识别。. CDRule 组件处理标准代码的转换，根据 requestName 来判断来源，且把源系统标准代码转换为MDM 标准

17、代码。CDRule 组件由 CDPool 从数据库中装载标准代码数据映射。映射关系是各个源系统指 向 MDM 。perites 文件是每个源系统接口中的需要转换的代码定义，其格式如下：CSCMQ=GenderType |HighestEducationType |TCRMAdminContEquivBObj.Admi nSystemType| TOLMQ=GenderType| 格式以 Key=Value 方式存放， key 为源系统提交的 requestName ，而值是以 | 分割 的多个数据域，每个数据域均需要进行标准代码替换。如果该数据域是于接口的更底层， 以 .

18、作为路径分割，如 TCRMAdminContEquivBObj.AdminSystemType 表示是于该对 象中的 TCRMAdminContEquivBObj 对象下的 AdminSystemType 需要进行代码转换。具体的实现类 CDRule 则根据输入 IBObj 对象和相应数据定义进行代码转换任务。RuleFW 是对外暴露的调用类，提供静态方法。* 规则调用处理 .*paramtype 规则类型，根据此类型确定是调用哪个规则实现 *paramobj 输入输出对象，规则实现对其进行处理，且形成结果返回 */publicstaticvoidcallRule(Stirngtype,IBO

19、bjobj)throwsIntegrateException; 壹期需要进行的代码转换为：个险GenderTypeM男M男F女F女U未知U未知TCRMPartyIdentificationBObj.IdentificationType0身份证00其它1参字第10后字第11空文字第12北文字13护照14装字第15北文字第RelationshipType1本人2丈夫3妻子4父亲5母亲6儿子HighestEducationType01文盲02 小学 03 初中 04 高中 05 大专 06 本科 07 研究生及之上08 中专 未知MaritalStatusType0未婚0未婚1已婚1已婚2离婚2离婚

20、3鳏寡3鳏寡AddressUsageTypeB单位地址2单位地址P邮递地址3邮递地址R家庭地址1家庭地址电商网上销售银保养老金团险. Utils 组件Utils 组件是工具类组件，主要包括服务调用组件，负责进行 MDM 服务的调用通过 IIOP 方式访问 MDM 的 EJB(DWLServiceController) 来处理 webservice 格式的 请 求 ， 具 体 请 参 考 MDMworkbench 中 的 。数据库访问. IntegException 组件例外处理的组件，如果处理过程中有例外，则需要记录例外的原因和状态，且把该服务 请求 xml 保存到

21、数据库。例外后需要保存的内容如下：错误流水号ERRLOGIDBigIntPK，自增型时间LOG_DTTimestamp错误返回消息ERRMESSAGEVarchar(255)exception.getMessage()或者业务逻辑错误说明，如找不到该客户等错误提交消息REQUESTXMLXml 字段错误堆栈同样需要记录到 log4j 日志中。批量数据集成(基于批量 /XML )批量处理的主要内容类似业务数据集成章节的近实时处理部分，主要是 FileParser 模块 管理各个业务系统上传的批量文件，且调用 MsgParser 模块解析具体的数据内容，同样， 根 据 规则 处 理 各个 业务逻

22、辑规则 ，然 后形 成 具体 的 批量 文 件 ， 且 调用 MDM 的 BatchProcessor 进行批量处理。具体模块关系如下图所示。批量处理的部分将会复用较多的近实时处理部分的业务组件，包括：MsgParser 组件IntegExceptionCDRule 组件组件2Constructor 读入文件，形成文件输入流和输出流，且调用批量处理，根据源系统类 型实例化具体的业务逻辑处理实例，传递输入流；3readOneService 从输入流读取壹个 service 块；4readOneService 返回读取的 service 块；5parse 调用 MsgParser 把 servie

23、 块的 xml 格式解析成为壹个 IBObj ；6parse 返回 IBObj ；7callRule 调用代码转换 rule ，转换客户证件类型、地址类型、联系类型为 MDM 标准 代码；8callRule 调用返回；9writeTmp 把客户关键信息写入数据库，内容参见临时表定义；10writeTmp 返回；之上从 3 到 10 循环处理，直到文件输入流到尾端；11resetFile 把文件输入流重新定位到开始；12resetFile 返回重新定位后的文件输入流；13callRule 调用批量客户识别的规则处理；13.1updateTmp 根据规则处理结果更新临时表，确定客户增加、修改类型以

24、及客户的 地址、联系、证件类型更改类型；13.2updateTmp 处理临时表结束；14callRule 调用规则返回；15readOneService 从文件输入流读入壹个 servicexml 块；16readOneService 返回 service 块；IBObj 的内容为标准进行批量处理， 完成后17parse 调用 MsgParser 解析为壹个 IBObj ；18parse 返回 IBObj ；19readTmp 读入壹条和 IBObj 匹配的临时表记录；20readTmp 读入返回数据；21callRule 调用格式转换规则处理，根据临时表类型标志，转换 服务的 IBObj ；

25、22callRule 返回转换完毕的 IBObj ；23toServiceXml 把IBObj 转换为标准的 xml 格式；24toServiceXml xml 结果；25writeXml2File 把xml 结果写入文件输出流；26writeXml2File 返回；之上 15 到 26 循环处理，直到文件处理完毕；27runbatch.sh 关闭相应资源， 调用 MDM 的 BatchProcessor 处理相应输入输出及中间文件；. 批量处理接口批量处理主要包括三个外部系统接口：银保增加客户养老金增加客户团险增加客户具体接口格式参见服务接口 外部系统输出符合接口规范的数据文件

26、，且 FTP 到规定的目录，具体参见服务接口中“批量接口模式”. FileParser 组件FileParser 组件是解析文件的处理，其主要处理客户的上传数据文件，同时新建输出结 果文件。解析整体文件格式；包括输入输出，输出文件每个 servicexml 只能是壹行；FileParser 调用 IntegrateFW 组件以处理不同的业务集成逻辑。FileParser 组件暴露壹个抽象类 IntegrateBatchLogic 供 IntegrateImpl 组件实现，其 抽象方法如下。* 执行具体的业务集成任务 .用于批量模式* 于 IntegrateImpl 组件实现该方法

27、，处理具体某壹项接口的逻辑调用，壹般包括* 代码转换、服务拆分、服务变换、 DSP 判断等*paramis 输入流，是通过 MQ 接收到的消息内容， xml 字节流*throwsItegrateException 例外，如果发生例外需要到例外组件处理*/ publicabstractvoidexecuteBatch(InputStreamis,OutputStreamos)throwsItegrateException;于 IntegrateBatchLogic 中仍实现必要方法：resetFile 重置文件指针到文件头；callRule 调用代码转换规则；getService 获取文件的下壹

28、个 service 块；callMDMBatch 调用 MDM 的 runbatch.sh. IntegrateImpl 组件 具体的业务集成逻辑实现，三个不同源系统的增加客户业务集成逻辑相同，其具体内容 格式有所差异。具体逻辑处理如下图。. 批量客户识别匹配规则 批量客户识别匹配规则是指考虑执行效率而进行的客户识别匹配程序，其匹配规则和于 线完全相同，但处理方式变成批量，规则参考如下：统壹客户管理平台进行客户识别的关键数据包括：客户名称+出生日期、证件类型 + 证件编号以及客户性别。 为了提高系统的客户识别能力和信息的准确度， 加快业务处理的速度， 系统针对用于客

29、户识别的关键数据进行了有效性规定，符合下列规定的信息才能够作为客户 识别的依据：客户名称不允许为空证件编号长度至少 8 位。 不满足上述有效性规则的客户信息会被加入统壹客户管理平台，但不会进行客户识别。 注：规则以于线匹配规则为准，请参考关联设计文档以确认规则壹致性； 客户匹配的主体是 MDM 数据库的 contact/person/identifier 表； 另外进行匹配的表是临时表： tmp_contactSql 参考：- 先根据客户号和源系统 id 进行匹配updatetmp_contacttset(MATCH_TP,MATCH_PARTYID,MATCH_PARTYUPDT,MATCH

30、_PERSONUPDT,PERSONNAMEID,PERSONNAME_UPDT)(select1,c.cont_id,c.last_update_dt,p.last_update_dt,pn.person_name_id,pn.last_update_dtfromcontactc,personp,tmp_contactt ,contequive,personnamepnwhere)whereexists(select1fromcontactc,personp,tmp_contactt,contequive ,personnamepnwhere)- 根据证件类型证件号码客户姓名客户生日进行匹配

31、updatetmp_contacttset(MATCH_TP,MATCH_PARTYID,MATCH_PARTYUPDT,MATCH_PERSONUPDT,PERSONNAMEID,PERSONNAME_UPDT(select1,c.cont_id,c.last_update_dt,p.last_update_dt,pn.person_name_id,pn.last_update_dtfromcontactc,personp,tmp_contactt,identifieri,personnamepnwherec.cont_id=p.cont_idandi.cont_id=c.cont_idan

32、dt.id_tp_cd=i.id_tp_cdandt.ref_num=i.ref_numandlength(rtrim(ltrim(t.ref_num)7 andt.person_nameisnotnullandt.person_name DER_TP_CODE=p.GENDER_TP_CODE)andchar(t.birth_dt,ISO)=substr(p.birth_dt,1,10)andt.match_tp=2andpn.cont_id=c.cont__usage_tp_cd=1)whereexists(selectc.cont_idfromcontactc,p

33、ersonp,tmp_contactt,identifieri,personnamepnwherec.cont_id=p.cont_idandi.cont_id=c.cont_idandt.id_tp_cd=i.id_tp_cdandt.ref_num=i.ref_numandlength(rtrim(ltrim(t.ref_num)7andt.contact_name=c.contact_nameandt.person_nameisnotnullandt.person_name ER_TP_CODE=p.GENDER_TP_CODE)andchar(t.birth_dt,ISO)=subst

34、r(p.birth_dt,1,10)andt.match_tp=2andpn.cont_id=c.cont__usage_tp_cd=1)- 根据匹配结果查找需要修改的证件号码(IDENT_MATCH_TP,IDENTIFIR_ID,IDENTIFIR_UPDT)=(select1,IDENTIFIER_ID,LAST_UPDATE_DTfromtmp_contactt,identifieriwheret.match_partyid=i.cont_idandt.id_tp_cd=i.id_tp_cdandt.ref_numi.ref_num)whereexists(se

35、lect1from,identifieriwheret.match_partyid=i.cont_idandt.id_tp_cd=i.id_tp_cdandt.ref_numi.ref_num);- 根据匹配结果查找需要修改的地址updatetmp_partyaddressset(ADDR_UP_TP,PARTYADDRESSID,PARTYADDRESSUPDT,LOCATIONGROUPUPDT)=(select2,a.location_group_id,a.last_update_dt,l.last_update_dtfromlocationgroupl,adderssgroupa,tm

36、p_contactt,tmp_partyaddresspawherel.location_group_id=a.location_group_idandl.cont_id=t.matcd_partyidandt.lob_tp=pa.lob_tpandt.lob_custno=pa.lob_custnoandpa.ADDR_USAGE_TP_CD=a.ADDR_USAGE_TP_CD)whereexists(select1fromlocationgroupl,adderssgroupa,tmp_contactt,tmp_partyaddresspawherel.location_group_id

37、=a.location_group_idandl.cont_id=t.matcd_partyidandt.lob_tp=pa.lob_tpandt.lob_custno=pa.lob_custno andpa.ADDR_USAGE_TP_CD=a.ADDR_USAGE_TP_CD);- 根据匹配结果查找需要修改的联系updatetmp_partycontactmethodset(CONT_METH_UP_TP,PARTYCONTMETHID,PARTYCONTMETHUPDT,LOCATIONGROUPUPDT)=(select2,a.location_group_id,a.last_upda

38、te_dt,l.last_update_dtfromlocationgroupl,contactmethodgroupa,tmp_partycontactmethodpawherel.location_group_id=a.location_group_idandl.cont_id=t.matcd_partyidandt.lob_tp=pa.lob_tpandt.lob_custno=pa.lob_custnoandpa.CONT_METH_TP_CD=a.CONT_METH_TP_CD)whereexists(select1fromlocationgroupl,adderssgroupa,t

39、mp_contactt,tmp_partyaddresspawherel.location_group_id=a.location_group_idandl.cont_id=t.matcd_partyid);. 临时表定义tmp_Contact源系统类型LOB_TPBigIntPK客户号LOB_CUSTNOVarchar(50)PK客户姓名PERSON_NAMEVarchar(50)客户证件类型ID_TP_CDBigInt客户证件号码REF_NUMVarchar(50)客户生日BIRTH_DTDate客户性别GENDER_TP_CODEBigInt匹配类型MATCH_TPInt1

40、 根据源系统客户 号匹配2 根据证件类型号 码等匹配， 3不匹配 缺省为 3匹配 partyIdMATCH PARTYIDBigint匹配 partyUpdateDTMATCH_PARTYUPDTTimestamp匹配personUpdateDTMATCH_PERSONUPDTTimestamp客户姓名 IDPERSONNAMEIDBigint客户姓名更新时间PERSONNAME_UPDTTimestamp证件更新标志IDENT_MATCH_TPInt1 更新证件2不更新缺省为 2证件 idIDENTIFIR_IDBigint证件更新时间IDENTIFIR_UPDTTimestamp需要考虑旧

41、客户的证件号码更新；tmp_PartyAddress源系统类型LOB_TPBigIntPK客户号LOB_CUSTNOVarchar(50)PK地址类型ADDR_USAGE_TP_CDBigIntPK地址修改类型ADDR_UP_TPInt1新增2更新缺省为 1客户地址 idPARTYADDRESSIDBigint客户地址 updatedtPARTYADDRESSUPDTTimestampLocationGroupupdtLOCATIONGROUPUPDTTimestamptmp_PartyContactMethod源系统类型LOB_TPBigIntPK客户号LOB_CUSTNOVarchar(5

42、0)PK联系类型CONT_METH_TP_CDBigIntPK联系修改类型CONT_METH_UP_TPInt1新增2更新缺省为 1客户联系 idPARTYCONTMETHIDBigint客户联系 updatedtPARTYCONTMETHUPDTTimestampLocationGroupupdtLOCATIONGROUPUPDTTimestamp灰色区域是原始写入客户信息，白色区域是匹配后的更新内容；1.1.3.批量型.客户匹配 1 遗漏未匹配客户处理遗漏未匹配客户是指于上壹个批量期间由于批量处理时间窗口内实时、近实时增加客户时和批量客户可能的合且情况，由于时间的差错导致形成 2 条不同记录的情况，同时这种情 况由于且发情况下不能完全锁定 DSP 处理的串行化从而可能出现的情况。遗漏未匹配客户处理是针对上述情况的客户合且处理，其条件为：新增客户 created_dt 为新增；符合匹配条件m/person.GEN