业务数据集成

1、业务数据集成业务数据集成是指 MDM Server 与外围系统之间的数据交换时需要中间处理的模块应用层。业务电商电商个险团险养老金银保总控自助服务网上销售HttpMQFTP(file)近实时处理批量处理业务集成层Ext Package数据服务层Base ServicesMDM infrastructure批量加载架构服务层MDM DB Module如图所示，外围系统要紧通过三种模式与 MDM Server 交互 (那个地点不考虑 ETL) ：直截了当的 web service访咨询 要紧通过 http 协议联机访咨询， 一样不通过业务集成层而直截了当访咨询数据服务层近实时的消息通知 要紧通过

2、MQ 消息访咨询，需要在业务集成层进行必要的处理批量处理 要紧通过文件进行数据传递，需要在业务集成层进行处理业务集成服务层的要紧内容包括：近实时的消息通知数据标准代码的转换交易 xml 的拆分MQ 组件是所有的客户信息更新消息的载体，MsgReceiver 组件负责接收 MQ 的消息，通过 MsgParser组件进行 xml 格式解析， MsgReceiver 转交到 IntegrateFW 组件，IntegrateFW 组件是转发框架， 负责业务集成层的差不多操纵，按照报文的类型调用必要的规则处理进行数据转换或者信息拆分，然后交由具体逻辑处理实现模块 IntegrateImpl 完成业务集成

3、逻辑，并最终通过 ServiceInvocation 模块调用 MDM 的服务，完成交易。数据服务层要紧包括扩展服务和组合服务。如上图，按照业务需求，一样性的顺序如下详述：1 ejbCreate MQ 中的信息通知 WAS 容器，按照 MsgReceiver 组件的 MDB 情形，创建一个实例进入池态；2 onMessage WAS 容器调用 MDB 的 onMessage方法，并把消息作为参数传递到该方法；2.1 deliver onMessage方法中调用 IntegrateFW 组件的 deliver 方法，并传递消息；2.1.1 parse IntegrateFW 组件调用 MsgPa

4、rser组件解析传入的消息体； Constructor MsgParser组件按照消息体解析，生成一个IBObj 对象；2.1.2 parse parse方法返回生成的IBObj 对象；2.1.3 getTCRMTxType 猎取 IBObj 对象的服务类型；2.1.4 getTCRMTxType 返回字符串结果；2.1.5 getRequestName 猎取 IBObj 的要求者；2.1.6 getRequestName 返回；2.1.7 Constructor IntegrateFW 按照服务对象和要求者， 实例化一个IntegrateImpl 组件中集成服务实例；2.1.8

5、 execute 调用具体实例的业务逻辑， 并把 IBObj 作为参数传递； callRule 调用标准代码转换规则，进行必要规则处理； callRule 返回规则处理后的IBObj 对象； matchRule 调用客户识不规则，进行识不 matchRule 调用返回 constructIBObj 在具体的业务逻辑处理中，按照具体需要调用 MsgParser创建需要的 IBObj 对象；.1 Constructor MsgParser构建新的 IBObj 实例； constructIBObj 返回

6、具体的 IBObj 实例； setAttribute 处理具体的 IBObj 实例内容； setAttribute 处理内容返回； toServiceXml 处理完所有的 IBObj 实例内容后，调用方法转换为 xml 标准服务格式；0 toServiceXml 返回 xml 字节流；1 invoke 调用 MDM 服务；2 invoke 返回；2.1.9 execute 完成具体业务集成逻辑实例的执行；调用；的调用需要包括在事务处理中；MsgReceiver 组件要紧由 IntegrateMDB 组成，Int

7、egrateMDB 是一个 Me ssageDriverBean，通过 onMessage(Message)方法接收 MQ 的消息通知。信息接收组件要紧负责从MQ 接收消息，参数Message使用 BytesMessage类型，消息格式为 XML 形式，其具体的接口定义参见 服务接口定义中的有关通过 MQ 接口部分。接口要紧包括：个险 增加客户 (addPerson / CSCMQ)个险 保全修改客户 (updatePerson / CSCMQ)电商网上销售 增加客户 (addPerson / TOLMQ)电商网上销售 修改客户差不多信息 (updatePersonBase/ TOLMQ)电商

8、网上销售 保全修改客户 (updatePerson / TOLMQ)注：以上括号中的服务名是指各个业务系统提交到 MQ 中的消息服务名，而不是 MDM 调用的标准服务名；组件调用 IntegrateFW 组件进行后续处理。public void onMessage(Messagemsg) if (msg instanceof BytesMessage) BytesMessage bm = (BytesMessage) msg; long length = bm.getBodyLength(); byte bs = new bytelength; bm.getBytes(bs);BytesInp

9、utStream bis = new BytesInputStream(bs);IntegrateFW.deliver(bis); else 件是框架 件， 集成 的操 度。 件 用然后按照 文 型 用具体的操 理 ，同 用 理 行数据和 理，最后形成 理后的 文提交 ，完成整个 理，同 如果在 理中 生 ， 交由如果 用出 需要 行回 理。/* 行具体的 集成任 . 用于 MQ 异步通知模式* 在 IntegrateImpl 件 方法， 理具体某一 接口的 用，一 包括* 代 、服 拆分、服 、 DSP 判定等* param is 输入流，是通过 MQ 接收到的消息内容， xml 字节流*

10、throws ItegrateException 例外，如果发生例外需要到例外组件处理*/public void execute(IBobj obj) throws ItegrateException;根据请求来源把消息开始中所有的标准代码转换为 MDM 代码根据客户号和 lob 类型查询 MDM 数据库 是否存在确定是否存在该客户组件对外调用的Fa? ade 接口调用是 IntegratFW 类，使用静态方法。否/*传递到 IntegrateFW 组件执行后续任务 . 用于 MQ 异步通知模式新增客户，调用调用在线可疑客户识否成功匹配*AddPerson服务别程序param is 输入流，是

11、通过MQ 接收到的消息内容， xml 字节流*/是是public static void根据已有deliver(InputStream客户名称调根据已is);有客户 Id更新用updatePersonName服务更新姓名根据客户号和 lob类IntegrateImpl开始 组件型查询 MDM 数据库确定是否存在该客户基本客户，调用updatePerson服务是否存在记录错误信息否例外处理IntegrateImpl 组件是按照报文类型确定的具体的业务集成逻辑的实现。根据客户号和 lob类记录错误信息根据已有 partyid 分拆根据已有 partyid 分拆形成开始否根据已有 partyid 分

12、拆型查询 MDM 数据库是否存在例外处理形成客户地址服务，客户电话、电子邮件联系是形成客户证件服务，一期需求要紧包括：确定是否存在该客户服务，调用调用调用个险增加客户结束changePartyAddress根据已有客户名称调用 updatePersonName服务更新姓名根据已有客户名称调用 updatePersonName服务更新姓名changePartyContactMethodchangePartyIdentifier根据已有客户 Id更新根据请求来源把消息基本客户，调用是中所有的标准代码转updatePerson服务换为 MDM 代码根据已有客户 Id更新根据请求来源把消息基本客户，调

13、用中所有的标准代码转updatePerson服务换为 MDM 代码个险修改客户差不多信息根据已有 partyid分拆结束形成客户地址服务，调用changePartyAddress个险保全修改客户信息根据已有 partyid分拆形成根据已有 partyid分拆客户电话、电子邮件联系形成客户证件服务，服务，调用调用changePartyContactMethodchangePartyIdentifier结束网上销售增加客户参见个险增加客户，只是其接口内容稍有差异网上销售修改客户差不多信息参见个险修改客户差不多信息，只是其接口内容稍有差异网上销售保全修改客户信息参见个险保全修改客户信息，只是其接口内

14、容稍有差异此组件需要处理地址、电话、电子邮件、客户标识等的格式变换，是把输入的对象格式变为标准的 MDM 扩展服务格式，包括：从粗粒度服务格式转换为updatePersonName服务格式转换为 changePartyAddress服务格式 服务中确定具体调用correctPartyAddress 依旧 addPartyAddress依旧不做处理；转换为 changePartyContactMethod服务格式 服务中定具体调用updatePartyContactMethod依旧 addPartyContactMethod依旧不做处理；从粗粒度服务格式转换为 changePartyIdenti

15、fer 服务格式 服务中定具体调用 updatePartyIdentifer依旧 addPartyIdentifer 依旧不做处理；具体接口参见服务接口和MDM 开发文档。MsgParser是暴露在外的 Fa? ade调用界面，通过两个个方法接收外部调用。/* 解析 xml.* param is 输入流，是通过 MQ 接收到的消息内容， xml 字节流*/public static IBObj parse(InputStream is);/* 把 IBObj 按照类型输出为标准的MDMservice 报文 .* param obj 输入的 IBObj* return 返回一个字节数组，组成xm

16、l，使用标准的iso-8859-1 格式*/public static byte toServiceXml(IBObj obj); /* 按照名称构建新的 IBObj 对象 .* param name 输入的 IBObj 名称* return 返回 IBObj 对象*/public static IBObj constructIBObj(String name);IBObj 是用于定义报文的bean，类似 MDM 中的 BObj。IBObj 要紧定义 xml 中的 bobj 对象，IBObj 内部能够嵌套。其暴露的接口如下。/* 设置 IBObj 的类型 . 是指对象类型，如 TCRMPers

17、onBObj、* TCRMAdminContEquivBObj 、TCRMPartyAddressBObj、TCRMAdd ressBObj 等。*/public void setType(String name);public String getType();/* 设置服务的类型 .*/public void setTCRMTxType(String type); pubic String getTCRMTxType();/* 设置服务的对象类型 .*/public void setTCRMTxObject(String obj);public String getTCRMTxObject

18、();/* 设置 IBObj 对象 .*/public void setIBObj(IBObj obj); /* 按照名字猎取 IBObj 对象 . 只能猎取下一级对象 .*/public IBObj getIBObj(String type); public String getAttribute(String type); /* 设置属性 . 自动区分处理 Extension 属性 .*/public String setAttribute(String type, String value); /* 设置服务头 .*/public void setHeader(String header

19、); public String getHeader();/* 设置要求操纵部分 .*/public void setRequestControl(String rc);public String getRequestControl();/* 设置要求操纵部分的要求名称 .*/public void setRequestName(String rn); public String getRequestName();/* 置要求操 部分的 LOB.*/public void setRequestLOB(String rlob); public String getRequestLOB();DSP

20、 Rule 件 理 DSP ，具体参 DSP 中在 可疑客 不。CD Rule 件 理 准代 的 ， 按照 requestName来判定来源，并把源系 准代 MDM 准代 。CDRule 件由 CDPool 从数据 中装 准代 数据映射。映射关系是各个源系 指向MDM 。perites文件是每个源系 接口中的需要 的代 定 ，其格式如下：CSCMQ= GenderType|HighestEducationType|TCRMAdminContEquivBObj. AdminSystemType|TOLMQ=GenderType|格式以 Key=Value 方式存放， key

21、 源系 提交的requestName，而 是以 分割的多个数据域，每个数据域都需要 行 准代 替 。如果 数据域是在接口的更底 ，以TCRMAdminContEquivBObj.AdminSystemTypeTCRMAdminContEquivBObj 象和相 数据定 行代 任 。RuleFW 是 外暴露的 用 ，提供静 方法。/* 规则调用处理 .* param type 规则类型，按照此类型确定是调用哪个规则实现* param obj 输入输出对象，规则实现对其进行处理，并形成结果返回*/public static void callRule(Stirng type, IBObj obj)

22、 throws IntegrateExc eption;一期需要进行的代码转换为：个险数据域源系统代码源系统值MDM 代码MDM 值GenderTypeM男M男F女F女U未知U未知TCRMPartyIdentificationBObj0身份证.IdentificationType00其它1参字第10后字第11空文字第12北文字13护照14装字第15北文字第RelationshipType1本人2夫君3妻子4父亲5母亲6亲小孩HighestEducationType01文盲 02小学 03初中 04高中 05大专 06本科 07 研 究生 及 以上 08中专 未知MaritalStatusTyp

23、e0未婚0未婚1已婚1已婚2离婚2离婚3鳏寡3鳏寡AddressUsageTypeB单位地址2单位地址P邮递地址3邮递地址R家庭地址1家庭地址电商网上销售银保养老金团险Utils 组件Utils 组件是工具类组件，要紧包括服务调用组件，负责进行MDM 服务的调用通过 IIOP 方式访咨询 MDM 的 EJB(DWLServiceController) 来处理 web service 格式的要求，具体请参考 MDM workbench 中的 com.ibm.mdm.traini ng.testerDWLServiceControllerTester。数据库访咨询IntegException 组件

24、例外处理的组件，如果处理过程中有例外，则需要记录例外的缘故和状态，并把该服务要求xml 储存到数据库。例外后需要储存的内容如下：字段代码类型备注错误流水号ERRLOGIDBigIntPK ，自增型时刻LOG_DTTimestamp错误返回消息ERRMESSAGEVarchar(255)exception.getMessage()或者业务逻辑错误讲明，如找不到该客户等错误提交消息REQUESTXMLXml 字段批量处理的要紧内容类似业务数据集成章节的近实时处理部分，要紧是 FileParserMsgParser模块解析具体的数据内容，同样，按照规则处理各个业务逻辑规则，然后形成具体的批量文件，并

25、调用批量处理的部分将会复用较多的近实时处理部分的业务组件，包括：MsgParser组件IntegException 组件CDRule 组件1 detectFile 守护线程，检查各个系统相应名目下的上穿文件是否ready；2 Constructor 读入文件，形成文件输入流和输出流， 并调用批量处理，按照源系统类型实例化具体的业务逻辑处理实例，传递输入流；3 readOneService 从输入流读取一个service 块；4 readOneService 返回读取的 service块；5 parse 调用 MsgParser把 servie块的 xml 格式解析成为一个 IBObj ；6 p

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

27、3.2 updateTmp 处理临时表终止；14 callRule 调用规则返回；15 readOneService 从文件输入流读入一个service xml 块；16 readOneService 返回 service块；17 parse 调用 MsgParser解析为一个 IBObj ；18 parse 返回 IBObj ；19 readTmp 读入一条与 IBObj 匹配的临时表记录；20 readTmp 读入返回数据；21 callRule 调用格式转换规则处理，按照临时表类型标志，转换 I BObj 的内容为标准服务的 IBObj ；22 callRule 返回转换完毕的IBObj

28、 ；23 toServiceXml 把 IBObj 转换为标准的 xml 格式；24 toServiceXml xml 结果；25 writeXml2File 把 xml 结果写入文件输出流；26 writeXml2File 返回；以上 15 到 26 循环处理，直到文件处理完毕；27 runbatch.sh 关闭相应资源，调用 MDM 的 BatchProcessor进行批量处理，完成后处理相应输入输出及中间文件；批量处理接口批量处理要紧包括三个外部系统接口：银保增加客户养老金增加客户团险增加客户具体接口格式参见服务接口到规定的名目，具体参见服务接口中“批量接口模式”组件是解析文件的处理，其

29、要紧处理客户的上传数据文件，同时新建输出结果文件。只能是一行；FileParser组件暴露一个抽象类 IntegrateBatchLogic 供 IntegrateImpl 组件实现，其抽象方法如下。/* 执行具体的业务集成任务 . 用于批量模式否* 在 IntegrateImpl 组件实现该方法，处理具体某一项接口的逻辑调用，循环到各个源系统对是否存在状态读入文件读入一个 service块一样包括是应目录检测上传文件文件* 代码转换、服务拆分、服务变换、 DSP 判定等* param is 输入流，是通过 MQ 接收到的消息内容， xml 字节流进行代码转换：证件把关键信息写入临时例外，如果

30、发生例外需要到例外组件处* throws ItegrateException类型、地址类型、联否文件结束表系类型理是*/从文件读入一个service块读入临时表信息一个重置文件到头批量客户匹配操作public abstract void executeBatch(InputStream is,OutputStream参见sq1参考 os) thro内存块ws ItegrateException;标准代码转换，把源根据已有 partyid分拆系统代码转换为形成客户地址服务，在 IntegrateBatchLogic 中还实现必要根方据已法有：客户名称形MDM 标准代码成updatePersonN

31、amecorrectPartyAddress格resetFile 重置文件指针到文件头；服务 xml 格式式或者addPartyAddress是callBatchSDP 调用批量客户识不规则；内存块结束修改客户，根据内存根据已有 partyid分拆形成客户电callRule 调用代码转换规则；话、电子邮件联系服务，否和文件内容，形成匹配updatePartyContactMethod格式或updatePerson服务 xmlgetService 猎取文件的下一个格式service 块；者 addPartyContactMethod客户匹配类型 调用 MDM的 runbatch.shcallMD

32、MBatch根据已有 partyid分拆否不匹配形成客户证件服务，把xml 输出到文件updatePartyIdentificati增加客户on或者IntegrateImpl 组件addPartyIdentification形成 addPerson服务xml 格式具体的业务集成逻辑实现，三个不同源系统的增加客户业务集成逻辑相同，其具体内容格式有所差异。是否结束调用 runbatch.sh具体逻辑处输理入如文件下备份图。临时文件清理批量客户识不匹配规则批量客户识不匹配规则是指考虑执行效率而进行的客户识不匹配程序，其匹配规则与在线完全相同，但处理方式变成批量，规则参考如下：统一客户治理平台进行客户

33、识不的关键数据包括：客户名称 +出生日期、证件类型 +证件编号以及客户性不。为了提升系统的客户识不能力和信息的准确度，加快业务处理的速度，系统针对用于客户识不的关键数据进行了有效性规定，符合下列规定的信息才能够作为客户识不的依据：客户名称不承诺为空证件编号长度至少8 位。不满足上述有效性规则的客户信息会被加入统一客户治理平台，但可不能进行客户识不。注：规则以在线匹配规则为准，请参考有关设计文档以确认规则一致性；客户匹配的主体是 MDM 数据库的 contact / person / identifier 表；另外进行匹配的表是临时表： tmp_contactSql 参考：- 先按照客户号和源系

34、统 id 进行匹配update tmp_contact t set(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_dtfromcontact c,person p,tmp_contact t,contequiv e,personname pnwherec.cont_id=p.cont_idand t.

35、lob_tp=e.ADMIN_SYS_TP_CDand t.lob_custno=e.ADMIN_CLIENT_IDand e.cont_id=p.cont_idand pn.cont_id=c.cont_idand _usage_tp_cd=1and (pn.end_dt is null or pn.end_dt current timestamp)and (c.end_dt is null or c.end_dt current timestamp)and (p.end_dt is null or p.end_dt current timestamp)and (e.end_d

36、t is null or e.end_dt current timestamp)where exists(select1fromcontact c,person p,tmp_contact t,contequiv e,personname pnwherec.cont_id=p.cont_idand t.lob_tp=e.ADMIN_SYS_TP_CDand t.lob_custno=e.ADMIN_CLIENT_IDand e.cont_id=p.cont_idand pn.cont_id=c.cont_idand _usage_tp_cd=1and (pn.end_dt is

37、null or pn.end_dt current timestamp)and (c.end_dt is null or c.end_dt current timestamp)and (p.end_dt is null or p.end_dt current timestamp)and (e.end_dt is null or e.end_dt current timestamp);- 按照证件类型证件号码客户姓名客户生日进行匹配update tmp_contact t set(MATCH_TP,MATCH_PARTYID,MATCH_PARTYUPDT,MATCH_PERSONUPDT,PE

38、RSONNAMEID,PERSONNAME_UPDT)=(select1,c.cont_id,c.last_update_dt,p.last_update_dt,pn.person_name_id,pn.last_update_dtfromcontact c,person p,tmp_contact t,identifier i,personname pnwherec.cont_id=p.cont_idand i.cont_id=c.cont_idand t.id_tp_cd=i.id_tp_cdand t.ref_num=i.ref_numand length(rtrim(ltrim(t.r

39、ef_num)7and t.contact_name=c.contact_nameand t.person_name is not nulland t.person_nameand (t.GENDER_TP_CODE is null and p. t.GENDER_TP_CODE is null) or t.GENDER_TP_CODE=p.GENDER_TP_CODE)and char(t.birth_dt, ISO)=substr(p.birth_dt, 1, 10)and t.match_tp=2and pn.cont_id=c.cont_idand _usage_tp_c

40、d=1and (pn.end_dt is null or pn.end_dt current timestamp)and (c.end_dt is null or c.end_dt current timestamp)and (p.end_dt is null or p.end_dt current timestamp)and (i.end_dt is null or i.end_dt current timestamp)where exists(select c.cont_idfromcontact c,person p,tmp_contact t,identifier i,personna

41、me pnwherec.cont_id=p.cont_idand i.cont_id=c.cont_idand t.id_tp_cd=i.id_tp_cdand t.ref_num=i.ref_numand length(rtrim(ltrim(t.ref_num)7and t.contact_name=c.contact_nameand t.person_name is not nulland t.person_nameand (t.GENDER_TP_CODE is null and p.GENDER_TP_CODE is null) or t.GENDER_TP_CODE=p.GENDE

42、R_TP_CODE) and char(t.birth_dt, ISO)=substr(p.birth_dt, 1, 10) and t.match_tp=2and pn.cont_id=c.cont_id and _usage_tp_cd=1and (pn.end_dt is null or pn.end_dt current timestamp) and (c.end_dt is null or c.end_dt current timestamp) and (p.end_dt is null or p.end_dt current timestamp) and (i.end

43、_dt is null or i.end_dt current timestamp);- 按照匹配结果查找需要修改的证件号码update tmp_contact t set(IDENT_MATCH_TP,IDENTIFIR_ID, IDENTIFIR_UPDT ) =(select1,IDENTIFIER_ID,LAST_UPDATE_DTfromtmp_contact t,identifier iwheret.match_partyid=i.cont_idand t.id_tp_cd=i.id_tp_cdand t.ref_numi.ref_numand (i.end_dt is null

44、or i.end_dt current timestamp)where exists(select1fromtmp_contact t,identifier iwheret.match_partyid=i.cont_idand t.id_tp_cd=i.id_tp_cdand t.ref_numi.ref_numand (i.end_dt is null or i.end_dt current timestamp);- 按照匹配结果查找需要修改的地址update tmp_partyaddress set(ADDR_UP_TP, PARTYADDRESSID, PARTYADDRESSUPDT,

45、 LOCATIONGROUPUPDT ) =(select2,a.location_group_id ,a.last_update_dt,l.last_update_dtfromlocationgroup l ,adderssgroup a ,tmp_contact t ,tmp_partyaddress pawhere l.location_group_id=a.location_group_id and l.cont_id=t.matcd_partyidand t.lob_tp=pa.lob_tpand t.lob_custno =pa.lob_custnoand pa.ADDR_USAGE_TP_CD=a.ADDR_USAGE_TP_CD and (l.end_dt is null or l.end_dt current timestamp)and (a.end_dt is null or a.end_dt current timestamp)where exists(select1fromlocationgroup l,adderssgroup a,tmp_contact t,tmp_partyaddress pawherel.location_group_id=a.location_