核算对账核心

清算对账系统

支付公司提供的所有金融服务是建立在银行资金体系之上的，支付公司账务系统内账户的资金都与其在银行的存款资金一一对应，为了保证真实的资金账户和虚拟账户的资金转换正确，支付公司必须及时与银行进行各类业务的资金核对，所有资金核对都依赖于银行的系统。

1. 资金流入与银行的对账

从银行流入的资金是由银行侧控制资金结转清算与对账时间，即每日客户通过银行向支付机构充值的资金是由银行实时通知支付机构充值指令的发生，银行在每日晚间经过汇总后向支付机构的银行收款账户入账，同时提供入账清算文件。支付机构获取该文件后，与业务数据进行核对。

对账结果若相符，则没有问题。若出现对账结果不符，很有可能是系统或者业务在某些环节发生问题，存在两种情况：

1. 银行充值明细多，支付机构充值明细少；即银行向支付机构入账资金多于支付机构业务发生情况，一般采取临时挂账处理，查出原因后再具体解决；

2. 银行充值明细少，支付充值明细多；即银行向支付机构入账资金少于支付机构业务发生情况，则可能对支付机构产生资金损失，一般采取临时挂账处理，查出原因后再具体解决。

2. 资金流出与银行的对账

从支付公司流出的资金是由银行侧控制资金结转清算与对账时间，即每日客户向支付机构申请提现的资金是在每日支付公司批量向银行发起提现请求时，从支付公司银行存款账户扣转，但扣转的结果一般需要一段时间才能从银行侧得到反馈。

当银行侧提供扣转成功和失败的清算文件，支付公司获取这些文件进行明细核对，对于提现失败的申请，由支付公司后台发起直接将资金回充入客户账户，不在对账中心进行对账处理。

什么是对账？

什么是资金对账？

在会计上的概念：指为了保证账簿记录的正确性而进行的有关账项的核对工作，做到账证相符、账账相符、账实相符。

在支付机构的概念：资金对账，在对账中心进行，将系统保存的账务流水与银行返回的清算流水和清算文件进行对账，核对系统账务数据与银行清算数据的一致性，保证支付机构各备付金银行账户每日的预计发生额与实际发生额一致。即核对银行实际清算资金如充值、充退、提现等业务的银行处理结果是否一致。

对账中心的作用？

是主要处理对账的系统模块，主要业务是清算对账。对账中心部署于工作平台，分别接受会计系统和清算系统的数据输入进行对账处理。

对账中心最主要的职责是勾兑银行清算流水与支付系统入账流水，用以检查反映银存实际账户的余额变化与支付系统内部户余额变化是否平衡。对于已经核对无误的银行清算流水和支付系统入账流水，分别进入相应的历史银行流水库和历史入账流水库。

对于勾兑结果中银行清算流水多于系统入账流水的，而操作人员不明确资金的来源，需要根据所设定的分类规则将暂不明确的资金进行挂账处理。而后我们认为该部分资金已经系统入账，可以入历史流水库。

因为此时的银存实际账户的余额增加，与之对应的是支付系统内部户余额也增加了，比如： T 日的挂账可能会在 T+N 日后进行销账确认，而后续的销账行为是对明细流水的业务分流处理，我们不应将 T+N 日的销账所产生的账务流水作为入账流水，不再需要到对账中心体现。

对账内容和数据来源

第一步

入账流水和清算流水进行核对，目的是保证对每一笔订单银行的处理结果和我们系统的业务处理结果一致。

第二步

对账汇总确认单和银行对账单进行核对，目的是通过轧余额等方式，核对各类业务的银行处理结果是否与银行实际清算给我们的资金一致。

1. 对账业务流程

对账业务流程图

2. 对账中心主要功能

银行发生资金变动的入账流水，包括：充值、提现、提现失败、充退、充退失败、退票、购汇、信贷放款、信贷还款、还贷失败一系列业务引起的账务变动信息。这些业务流水在账务系统入账后，会计核心接收到登记分录请求并处理完毕，发送该流水至对账中心，对账中心对于某个业务的反向资金变动所产生入账流水也同步至对账中心。

这样做的目的是让待清算与入账流水在日切点保持等额，比如：提现 T 日会员申请提现，生成文件后会员账户提现金额转入待清算提现款项，与此同时发送流水到对账中心，此时待清算与对账中心入账流水是保持平衡的。

而 T+1 日银行处理失败，系统会回充带清算提现款，如果此时不发送失败流水到对账中心与其对应的流水进行购销的话，则入账流水就会不变，和带清算提现不平衡。

Tips：

1. 完整的日结流程支持：银行清算流水入库 → 流水对账 → 流水分类 → 汇总确认单 → 自动挂账 → 销账确认 → 操作员轧账 → 总账日结以及登记银行余额等。

2. 完善的报表模型输出：如按入账日期 / 银行日期 / 清算日期等的统计报表、银行账户余额报表、未达款项报表等；

3. 提供日切服务：在经确认后进入历史清算流水的，同步汇总该部分数据进入历史流水汇总表中保存，所以在日切时，只需直接将该汇总数据再次汇总即可。

4. 外围业务功能支持：提供对于银行账户（ 独立于对账中心之外，不参与处理逻辑 ）的管理功能，支持与内部户的映射管理，用以满足部分报表需求；提供基于账务核心所提供的业务代码查询功能；提供对于财务系统的交互支持，包括与财务科目的对应关系管理、通知流水数据等；提供用以校验订单与清算流水匹配状态的错账核实功能。

对账流程图

对账中心功能模块分析

获取资金对账数据：

• 获取方式：需要进行清算流水导入的文件一般通过人工在页面上导入，另外有些业务的流水文件通过系统自动匹配或者定时获取的方式得到。

• 清算流水导入渠道非常多：需要统一各清算流水导入功能到一个页面入口， 同时引入清算通道对账的逻辑，将清算流水导入过程需要映射清算通道（ 包含新增清算流水等 ）。

• 各类业务的清算流水文件的解析和导入逻辑不一样。

清算流水对账：

对账逻辑：

一对一对账：入账流水只可能存在一条，银行入账流水也仅存在一条，然后一对一去对。目前按照一对一对账的业务涉及到：网银 B2C 充值、网银 B2B 充值、VISA、网汇E、卡通充值、正常提现、认证提现、银企互联提现、卡通提现、卖家信贷、信用卡还款提现、COD、网点支付。

对账标准：清算通道 + 订单号 + 金额 Or 银行名称 + 业务代码 + 订单号 + 金额。

满足一对一的业务如下：

• 提现成功：500401

• 认证提现：500402

• 还贷：500404

• 卡通提现：500403

• 个人网银充值：400301

• 企业网银充值：400302

• VISA：400314

• 网汇e：400315

• 卡通充值：400304

• 贷款：400307

• 银企互联提现：500405

• 信用卡还款提现：500407

• 后台强制提现500406

• COD

• 网点支付

Tips：

• 清算流水有，入账流水也有，金额一致，对账成功，清算流水、对账流水打上 ‘对账成功’ 标志，记录对账日期为系统当日；

• 清算流水有，入账流水也有，金额不一致，清算流水记金额不等，记录对账日期为系统当日，入账流水不变；

• 清算流水有，入账流水该订单号没有，清算流水打上 ‘银行多帐’ 的标志，记录对账日期为系统当日；

• 清算流水有，入账流水没有该业务代码初始状态的，清算流水打上 ‘银行多帐’ 的标志，记录对账日期为系统当日；

• 清算流水有，入账流水没有该银行的初始状态的记录，清算流水打上 ‘银行多帐’ 的标志，记录对账日期为系统当日；

• 对账之前需要判断 1 对 1 的流水中是否有重复，有重复的返回失败不进行后续对账。

多对多对账：入账流水里相同订单号，相同清算渠道，相同，金额相同业务代码的流水存在对条（比如充退）；而银行清算流水也可能是存在多条的情况的，这种情况下是多对多去对账的，遵循先到的先对的原则。

对账标准：清算通道 + 订单号 + 金额 Or 银行名称 + 业务代码 + 订单号 + 金额。

满足多对多的业务如下：

• 充退：410401

• 银企互联充退：410402

• Motopay 充退：410403

Tips：

• 清算流水有，入账流水有，满足对账标准，则对账成功，清算流水、对账流水打上 ‘对账成功’ 标志，记录对账日期为系统当日；

• 清算流水有，入账流水有，金额不等，清算流水打上 ‘金额不等’ 的标志，记录对账日期为系统当日，入账流水不变；

• 清算流水有，入账流水没有该订单号，清算流水打上‘银行多帐’ 的标志，记录对账日期为系统当日；

• 清算流水有，入账流水没有初始状态 410401 的流水，清算流水打上 ‘银行多帐’ 的标志，记录对账日期为系统当日；

• 清算流水有，入账流水没有这个银行初始状态 410401 的流水，清算流水打上 ‘银行多帐’ 的标志，记录对账日期为系统当日；

• 清算流水有，入账流水没有初始状态的 ‘410401’ 的流水，清算流水打上 ‘银行多帐’ 的标志，记录对账日期为系统当日。

一对多对账：入账流水有多条，和银行的一条去对账。

对账标准：清算通道 + 订单号 + 金额 Or 银行名称 + 业务代码 + 订单号 + 金额

涉及的业务：境外收单

• 境外收单购汇扣款（520101）

• 购汇益回补购汇账户（400319）

Tips：

• 入账流水存在 2 条 520101 的，清算流水存在 1 条 520101 的，将入账流水的 2 条加和后的金额和清算流水的进行对账，一致的为对账成功；

• 入账流水存在 1 条 520101 的，1 条 400319 的，那么要将 2 条金额之差和清算流水的 1 条 520101 的进行对账。不会出现 3 条 520101 或者 2 条 400319 的情况。

内部流水购销 ：内部流水购销是针对那种有入账流水表来说的，比如提现业务，提现文件生成就会扣款此时会同步一笔入账流水；而回导失败之后又会回充，此时也会同步一笔反向的业务流水，这两条流水不用再去和银行的资金流水进行对账，直接在入账流水这边进行购销即可。

需要购销的业务如下：

• 购销的规则：非已勾销状态下，同一银行同一订单号，金额一致而业务代码相反的正向流水进行勾销，而且一条反向流水只勾销一条；

• 日终轧账时候的购销：日终轧账的时候会对未购销的流水再次进行购销确保当天的流水都购销完全。（这个情况是为了解决反向流水先于正向流水先出现的逻辑错误情况而加的并提示流水号）；

• 如果业务代码相反，而金额不等的情况，就无法进行购销，这种情况原来是作为违反逻辑规则来进行标记的。这部分数据进行查明之后，可以修改流水之后，在日终轧账的时候从新进行购销；

• 购销分 ‘一对一’（提现）和 ‘多对多’ （充退）的购销；

• 勾销的流水不入历史库。

对账汇总确认单：

显示对账结果，选择后续的相应的处理界面，复核员在此模块进行流水的确认，还有的功能就是对于已经对账处理的银行清算流水与系统入账流水进行后续业务分流推进。

• 对于大部分对账成功的数据，可以将这些流水确认，确认后该部分流水将进入历史清算流水，只有进入历史清算流水的才认为银行与支付系统数据对平。对账成功的银行清算流水、账务入账流水全部进入各自历史清算流水表保存，相应的银行清算流水与入账流水同步删除。

• 金额不等的流水，可能是银行清算流水文件有误，也可能是错账，采取的做法是操作员手工修改金额，在银行清算流水管理里操作或直接删除，将其推进到下一轮的对账处理环节，对账是可以重复对的，只要满足初始状态要求的流水即可。

• 多账部分，由于各种已明确/未明确的原因，银行已经实际清算给支付公司了，但支付公司的账务系统并没有入账，需要提供一个流水分类和分类汇总挂账的操作入口。系统默认多账的流水给出的菜单是流水分类，金额不等的流水给出的流水管理页面。区别在于，在点流水分类时，只允许修改业务类型、银行日期，而且改完后不改变多账状态；流水管理，修改状态后，流水的对账状态将变成初始状态，需要进行重新对账。

• 流水进入历史库的校验规则。

举例说明：A 操作员导入了一笔流水，系统对账成功，银行清算流水与账务入账流水状态都被置为成功状态；B 操作员再次导入该相同流水，由于入账流水处于成功状态的可以继续对账，所以再次对账成功。系统在对账环节认为正常，但在入历史清算流水时必须做每组流水数据的平衡检查：必须是清算流水总额与入账流水总额匹配才可以进历史数据。

对于上述的场景，如复核员针对 B 操作员的对账成功流水确认后，可以正常进入历史流水，对账批次号认最后操作的批次。而继续对 A 操作员的成功对账流水确认，则系统将认为是不合法的入历史流水行为。或者不对特定操作员的汇总确认，系统必须检验出所存在的重复银行清算流水，并将对应的明细数据显示复核员。此时可将该重复对账的清算流水删除即可。

流水分类和分类汇总挂账：

• 对于多账的流水对账中心提供一个单独的处理入口，首先在此处进行分类，然后进行汇总挂账处理。操作人员可以根据多账流水的未确认类型修改成对应的业务代码，允许修改成业务类型为 7 的可挂账的类型（其他业务代码不允许挂账）；流水可以重复分类，系统不做控制；但已申请挂账的除外，分类完毕的多账流水可直接提供分类汇总挂账操作。

• 系统根据既定的业务代码判断是待查收入或待查支出挂账，页面跳转至相关凭证登记页面，此处业务规则同待查收支挂账；另一部分比如退票，因为不经过对账环节，所以需要直接在清算流水查询里面新增，业务代码 700106 未确认退票，不需要经过对账，直接去汇总确认单里将流水挂账。允许挂账的业务代码如下 ：700101 其他应付款，700102 其他应收款，700103 银行错账，700104 未确认结汇款，700105 线下汇款，700106 未确认退票，700107 未确认收款，700108 未确认支出款。

• 提交凭证登记申请相关校验：流水此刻状态是否是 ‘多帐’ ，申请提交人是否是当时的对账人，凭证登记成功，清算流水记录凭证号，流水状态改为 ‘已挂账’ ；

• 汇总挂账的反交易同步入账流水，金额是负值，系统在清算流水里在做一条数据，自动对账。

清算/入账/历史流水管理：

清算流水管理

对于所有的清算流水，都可以在此模块下进行查询、修改、删除，同时也可以新增流水。

• 查询功能，银行名称、业务代码、银行日期不能为空，业务代码和银行可以多选。

• 修改功能，处于未对账（初始）、金额不等、多账、对账成功状态的流水可以进行单笔、批量修改操作，这里的批量修改动作与流水分类功能类似，只是该功能入口不仅支持对批量流水的业务代码、银行日期更新，也支持其他要素信息批量修改，作为提供给操作员在未对账前批量修改已知流水的手段之一；对已挂账的流水，不允许更新。

• 流水删除：处于未对账（初始）、金额不等、多账、对账成功状态的流水可以进行单笔、批量删除操作，已挂账状态不允许删除。

入账流水管理

入账流水管理功能提供对账务入账流水的查询以及下载功能；为了杜绝对入账流水的人为变更操作，禁止支持对入账流水进行修改或删除处理。

• 查询时银行名称、业务代码、入账日期不能为空，业务系统流水号：针对某类业务的业务流水号，如充值业务就是充值流水号；账务流水号：账务处理的流水号，即账务操作记录数据的序号，在账户明细查询里可以获取。

• 对于提现类失败回充的流水，在操作员进行对账动作后，系统自动进行购销，在入账流水查询时，将对账状态选为已购销可以查到，这些不进入历史库，根据数据量，系统定时清理这些数据。

历史流水管理

分为历史入账流水和历史清算流水，查询功能在同一页面，查询时可以一起查询，也可以单独查询。历史流水只供查询和下载，不允许进行修改和删除操作。选择历史银行流水和历史清算流水（清算日期的时间间隔不能超过 7 天）进行查询，历史入账流水中无银行日期，历史银行流水中无入账日期。

银行余额录入功能：

在该模块下，实现对每个银行账户的实际余额录入，以此来和内部账户余额进行匹配校验。分为银行余额登记，银行余额导入（复核），以及银行余额查询功能。

银行余额登记：选择对应日期以及银行账户录入后，保存，此时去银行余额查询是看不到录入余额的，需要复核导入核对完毕后才能看到，保证核对的有效性。登记后系统记录一个临时余额。系统每天凌晨 1:15 的时候会跑批，取上一日已复核过的余额自动带入下日临时余额，如果不登记的话，就取上日余额。

银行余额导入：标准格式是一个填写银行余额的 excle 模板，导入后的核对逻辑，对复核员导入的数据进行检测该帐户是否有临时余额。根据银行账号、银行日期、银行余额等条件查询银行余额表。

判断 1 ：如果查询出结果为空那么将返回给用户：“这个银行帐号找不到对应的银行临时余额！”。

判断 2 ：如果查询结果大余一条，将返回给用户：“这个银行帐号的一天有多个银行临时余额，不正常！”

判断 3 ：检测该银行帐号对应的银行余额是否相等。如果不相等将返回给用户：“临时余额和实际余额不相等，核对失败！”当余额核对成功后，将复核员导入的余额写入该对应帐户的实际余额，并修改余额状态为已复核。实际余额更新成功后，将删除当天日期的临时余额。

银行余额查询：对操作员登记余额的信息，以及复核员复核过的余额进行查询，查询条件：银行名称、银行帐户、帐户状态、银行日期、余额状态。帐户状态：分为正常、废除、和销户三个状态，默认为正常状态。余额状态：分为未录入（N）、已录入（A）、已复核（Y）三个状态，默认为全部。

内部账户登账：

业务简述

针对日常结算工作中非银行待查类的内部账户进行登记，如：清算款、利息、手续费等；该业务登记不产生后续业务登记行为，即不具有作为初始凭证号的使用功能。对于批量销帐类业务处理中，多会员转帐失败的，导致过渡户上有剩余金额情况的，可通过此处进行内部户登账，将过渡户（负债）和银存（资产）余额同时减少，再通过待查收入挂帐实现平衡。

业务校验规则

1. 必须登记的是一借一贷帐户信息；

2. 借贷方帐户不能一致；

3. 金额必须大于 0 ；

4. 不允许任意一方是销帐帐户。

是否同步对帐中心流水：不需要同步对帐中心流水。

待查收入挂账：

业务简述

应用于结算操作员针对日常结算工作中的银行待查收入进行登记，所产生的业务凭证号作为后续销帐业务的原业务凭证号，并且作为所有该登记所引发的后续业务登记的初始凭证号。其中待查收入挂帐业务凭证的借方（银存帐户）所对应的银行名称作为后续销帐业务的银行名称，包括差额重新挂帐部分再销帐的业务凭证，都递沿该银行名称。

业务校验规则：

1. 必须登记的是一借一贷帐户信息；

2. 借贷方帐户不能一致；

3. 金额必须大于 0 ；

4. 借方帐户必须是银存帐户；

5. 贷方帐户必须是销帐帐户。

是否同步对帐中心流水：不需要同步对帐中心流水。

待查收入确认：

业务简述

针对银行待查收入登账的挂帐，可以通过本模块进行销帐。此处采取销帐确认方式进行处理，需要选择相应的待查收入挂帐业务凭证进行销帐业务登记。该业务登记可能产生后续业务登记行为，如差额销帐情况下，系统会自动做不足额部分的重新挂帐并复核通过，所产生的挂帐凭证作为后续销帐凭证的销帐卡片号。

业务校验规则：

1. 所销的原待查收入挂帐凭证必须合法；

2. 所销的原待查收入挂帐凭证必须已复核通过，处理完毕；

3. 所销的原待查收入挂帐凭证不能已被销帐；

4. 销帐总额(贷方发生额之和不得大于原待查收入挂帐凭证发生额)并大于 0 ；

5. 贷方必须是内部户。

待查支出挂账：

业务简述：

针对日常结算工作中的银行待查支出进行登记，所产生的业务凭证号作为后续销帐业务的原业务凭证号，并且作为所有该登记所引发的后续业务登记的初始凭证号。其中待查收入挂帐业务凭证的贷方（ 银存帐户）所对应的银行名称作为后续销帐业务的银行名称，包括差额重新挂帐部分再销帐的业务凭证，都递沿该银行名称。

业务校验规则：

1. 必须登记的是一借一贷帐户信息；

2. 借贷方帐户不能一致；

3. 金额必须大于 0 ；

4. 贷方帐户必须是银存帐户；

5. 借方帐户必须是销帐帐户 。

是否同步对帐中心流水：不需要同步对帐中心流水。

待查支出确认：

业务简述

针对银行待查支出登账的挂帐，可以通过本模块进行销帐。此处采取销帐确认方式进行处理，需要选择相应的待查支出挂帐业务凭证进行销帐业务登记。该业务登记可能产生后续业务登记行为，如差额销帐情况下系统会自动做不足额部分的重新挂帐并复核通过，所产生的挂帐凭证作为后续销帐凭证的销帐卡片号。

业务校验规则：

1. 所销的原待查支出挂帐凭证必须合法；

2. 所销的原待查支出挂帐凭证必须已复核通过，处理完毕；

3. 所销的原待查支出挂帐凭证不能已被销帐；

4. 销帐总额(借方发生额之和不得大于原待查收入挂帐凭证发生额)并大于0 ；

5. 借方必须是内部户。

意外数据恢复逻辑1. 意外数据恢复逻辑

重复支付：对同一内部订单号进行了二次或二次以上的支付。

支付失败，金额不等：买家实际支付的金额与交易金额不等。一般产生的原因是，买家在支付时，产生了掉单，卖家随后修改了交易价格。 在进行网银对账的时候，即会出现订单金额和交易金额不等的情况，且是一笔掉单。2、3 两类情况只发生在支付上。

支付成功，金额不等（这一类异常，偶尔有发生）：商户订单状态为成功，后台订单状态也为成功，并且交易状态是买家已付款，等待卖家发货。

（金额不等，支付成功，是因为会员对一个交易进行支付，但由于网络或银行系统等原因支付公司未接收到银行扣款信息，支付侧交易状态未予以变更，后卖家对该交易修改了价格，买家又对该修改过的价格进行了支付。但该支付成功的信息仍然没有被支付侧接收到，该交易状态仍未变更，后支付侧后台人员先对后面的那笔意外数据进行了恢复，后再对前面那笔意外数据恢复，就会出现这种“金额不等，支付成功”的数据）

2. 对帐及异常恢复逻辑

以商户成功订单为准：

用商户上的成功订单与后台的订单来核对：

• 若商户订单为成功，后台为初始或者失败，则更改后台状态为成功；

• 若后台为成功，商户成功订单中无该订单（时间差），则不更改后台状态；

• 若后台为初始或失败，商户成功订单中无该订单，则不更改后台状态。

不重复恢复：

T+1 日恢复T 日的订单，并且在 T+1 日后不再下载 T 日的订单进行二次或多次恢复。（为考虑会员感受，T 日下班前恢复 T 日 0 点到下班时点的订单）

时间性差异：

由于各家银行系统日切点均不同，并且大多不会在每日的 24 点（或早或晚），所以下载到的 T 日的订单流水与后台T日（ 0:00－24:00 ）的订单流水并不能全部对应上。将商户订单流水与后台订单流水核对，会出现商户有，后台无；商户无，后台有；商户有，后台有三种情况。对于 1、2 两种情况为我们所说的时间性差异。

商户数据与后台数据的关系为：

商户数据－（商户有，后台无）＋（商户无，后台有）＝ 后台数据

基于千万数据的在线表结构修改

　  在系统正常运作一定时间后，随着市场、产品汪的需求不断变更，比较大的一些表结构面临不得不增加字段的方式来扩充满足业务需求；并且是不能停止服务器。在 MySQL 在体量上了千万、数据的时候，Alter Table 的操作，可以让你等一天，而且在高峰期执行这种 SQL ，会造成表被锁，增加字段的操作需要很长的时间 ，让你的数据库也承担着压力。 

方案： pt-online-schema-change 工具

       https://segmentfault.com/a/1190000014924677

方案二：土方法 rename;

1.       创建目的表结构的空表，A_new (新加的字段也有了);

2.       在A表上创建触发器，包括增、删、改触发器;这样在copy的过程中,有其他的操作的话可以同步进行处理;

3.       通过insert…select…limit N 语句分片拷贝数据到目的表 (这个可以通过一个定时器Job)去做这些事情。

4.       Copy完成后，将A_new表rename到A表。

触发器例子：

增加触发触发器;

修改触发触发器:

删除触发触发器:

基于binlog 增量的数据解析服务

什么是binlog

binlog是mysql的一种二进制日志文件，用来记录数据的变化。mysql使用binlog进行主从复制，如图：

客户端向master的mysql sever写入数据

1. 当数据发生变化时，master将变更的数据记录写入到二进制文件中，即binlog。

2. slave订阅了master的binlog，所以会通过一个I/O THREAD与master的DUMP THREAD进行通信，同步binlog

3. I/O      THREAD读取到binlog后会吸入到relay log中，准备重放。

4. slave会通过SQL THREAD读取relay log，重放数据的改动并执行相应的改动。

这里有几点需要注意：

1. 主从复制不是强一致性，只能保证最终一致

2. master配合binlog复制会影响性能，所以尽量不要在master上挂太多的slave，如果对时间要求不高，可以在slave上挂slave

2.binlog的业务应用

  上面介绍了mysql中应用binlog的场景，而我们的业务可以伪装成master的slave节点，感知数据的变化，这就给了我们很多的业务运用空间。

2.1 数据异构

 经常有这样一个场景：

  原来业务是一个很单一的系统，所以表也在一起。随着业务的发展，系统开始拆分，总有一些表是各个业务都关注的表，但是对相关的字段的运用场景不同，所以这样一份元数据怎样更好的为各个系统服务就成了问题。当然，多写或者读写分离可以从物理节点上减少对数据服务器的压力，但是对业务并没有做到足够的支持，因为这些表都是一样的。因此我们可以通过binlog进行数据异构。

如图所示，订单系统生成订单后，通过binlog可以解析生成用户维度的订单信息供用户中心查询、商户维度订单表供运营管理，以及搜索系统的搜索数据，提供全文搜索功能。

这样，我们就通过原始的订单数据异构到三个系统中，提供了丰富的数据访问功能。不仅从节点上降低了数据服务器的压力，数据表现形式也更贴近自己的服务，减少不必要的字段冗余。

2.2 缓存数据的补充

对于高并发的系统，数据库往往是系统性能的瓶颈，毕竟IO响应速度是远远小于电子的运算速度的。因此，很多查询类服务都会在CPU与数据库之间加上一层缓存。即现从缓存获取，命中后直接返回，否则从DB中获取并存入缓存后返回。而如果原始数据变化了但缓存尚未超时，则缓存中的数据就是过时的数据了。当数据有变更的时候主动修改缓存数据。

当客户端更改了数据之后，中间件系统通过binlog获得数据变更，并同步到缓存中。这样就保证了缓存中数据有效性，减少了对数据库的调用，从而提高整体性能。

2.3 基于数据的任务分发

有这样一个场景：

  很多系统依赖同一块重要数据，当这些数据发生变化的时候，需要调用其他相关系统的通知接口同步数据变化，或者mq消息告知变化并等待其主动同步。这两种情况都对原始系统造成了侵入，原始系统改一块数据，并不想做这么多其他的事情。所以这时候可以通过binlog进行任务分发。

当原始业务系统修改数据后，不需要进行其他的业务关联。由调度系统读取binlog进行相应的任务分发、消息发送以及同步其他业务状态。这样可以将其他业务与原始业务系统解耦，并从数据的角度将所有管理功能放在了同一个调度系统中，责任清晰。

2.4 可以用于数据平滑迁移:

   https://www.w3cschool.cn/architectroad/architectroad-data-smooth-migration.html

2.5 数据抽取：

https://blog.csdn.net/u010670689/article/details/81066807

mysql bin-log相关比较好的开源项目

https://blog.csdn.net/everlasting_188/article/details/53304530

binlog的业务实践:

1.  首先需要开启mysql bin_log功能;

启动成功之后，我们可以登陆查看我们的配置是否起作用

show variables like '%log_bin%'

本次解析只考虑insert、update、delete三种事件类型

采用OpenReplicator解析MySQL binlog

Open Replicator是一个用Java编写的MySQL binlog分析程序。Open Replicator 首先连接到MySQL（就像一个普通的MySQL Slave一样），然后接收和分析binlog，最终将分析得出的binlog events以回调的方式通知应用。Open Replicator可以被应用到MySQL数据变化的实时推送，多Master到单Slave的数据同步等多种应用场景。Open Replicator目前只支持MySQL5.0及以上版本。

Open Replicator项目地址：https://github.com/whitesock/open-replicator

binlog事件分析结构图

在阅读下面的内容时，首先需要对binlog有一定的了解，可以 参考《MySQL Binlog解析》。

这里通过open-replicator解析binlog日志事件（binlog-format = row）。binlog日志事件里存在两种操作：DDL和DML，当DDL时输出一条sql，当DML时输出相关行信息。可以参考下面：

DDL（CREATE, ALTER, DROP, TRUNCATE，主要用在定义或改变表的结构）:

{

  "eventId": 1,

  "databaseName": "canal_test",

  "tableName": "`company`",

  "eventType": 2,

  "timestamp": 1477033198000,

  "timestampReceipt": 1477033248780,

  "binlogName": "mysql-bin.000006",

  "position": 353,

  "nextPostion": 468,

  "serverId": 2,

  "before": null,

  "after": null,

  "isDdl": true,

  "sql": "DROP TABLE `company` /* generated by server */"

}

DML(SELECT, UPDATE, INSERT, DELETE，对数据库里的数据进行操作):

{

  "eventId": 0,

  "databaseName": "canal_test",

  "tableName": "person",

  "eventType": 24,

  "timestamp": 1477030734000,

  "timestampReceipt": 1477032161988,

  "binlogName": "mysql-bin.000006",

  "position": 242,

  "nextPostion": 326,

  "serverId": 2,

  "before": {

    "id": "3",

    "sex": "f",

    "address": "shanghai",

    "age": "23",

    "name": "zzh3"

  },

  "after": {

    "id": "3",

    "sex": "m",

    "address": "shanghai",

    "age": "23",

    "name": "zzh3"

  },

  "isDdl": false,

  "sql": null

}

相关的类文件如下： 

OpenReplicatorTest：

public class OpenReplicatorTest {
 
     private static final Logger logger = LoggerFactory.getLogger(OpenReplicatorTest.class);
     private static final String host = "192.168.56.101";
     private static final int port = 3306;
     private static final String user = "root";
     private static final String password = "123456";
 
 
     public static void main(String[] args) throws IOException {
 
 
         OpenReplicator or = new OpenReplicator ();
         or.setUser(user);
         or.setPassword(password);
         or.setHost(host);
         or.setPort(port);
         MysqlConnection.setConnection(host, port, user, password);
 
         //      or.setServerId(MysqlConnection.getServerId());
         //配置里的serverId是open-replicator(作为一个slave)的id,不是master的serverId
 
         BinlogMasterStatus bms = MysqlConnection.getBinlogMasterStatus();
         or.setBinlogFileName(bms.getBinlogName());
         or.setBinlogPosition(4);
         or.setBinlogEventListener(new NotificationListener());
         try {
             or.start();
         } catch (Exception e) {
             logger.error(e.getMessage(),e);
         }
         Thread thread = new Thread(new PrintLogEvent());
         thread.start();
     }
 
     public static class PrintLogEvent implements Runnable{
         @Override
         public void run() {
             while(true){
                 if(CDCEventManager.queue.isEmpty() == false)
                 {
                     LogEvent ce = CDCEventManager.queue.pollFirst();
                     String prettyStr1 = JSON.toJSONString(ce);
                     System.out.println(prettyStr1);
                 }
                 else{
                     try {
                         TimeUnit.SECONDS.sleep(1);
                     } catch (InterruptedException e) {
                         e.printStackTrace();
                     }
                 }
             }
         }
     }
 
 
 }

LogEvent：

public class LogEvent implements Serializable {
 
     /**
      * 只针对delete、insert、update事件
      */
     private static final long serialVersionUID = 5503152746318421290L;
 
     private long eventId = 0;//事件唯一标识
     private String databaseName = null;
     private String tableName = null;
     private int eventType = 0;//事件类型
     private long timestamp = 0;//事件发生的时间戳[MySQL服务器的时间]
     private long timestampReceipt = 0;//Open-replicator接收到的时间戳[CDC执行的时间戳]
     private long position = 0;
     private long nextPostion = 0;
     private long serverId = 0;
     private Map<String, String> before = null;
     private Map<String, String> after = null;
     private Boolean isDdl = null;
     private String sql = null;
 
     private static AtomicLong uuid = new AtomicLong(0);
 
     public LogEvent() {}
 
     public LogEvent(final AbstractBinlogEventV4 are, String databaseName, String tableName) {
         this.init(are);
         this.databaseName = databaseName;
         this.tableName = tableName;
     }
 
     private void init(final BinlogEventV4 be) {
         this.eventId = uuid.getAndAdd(1);
         BinlogEventV4Header header = be.getHeader();
         this.timestamp = header.getTimestamp();
         this.eventType = header.getEventType();
         this.serverId = header.getServerId();
         this.timestampReceipt = header.getTimestampOfReceipt();
         this.position = header.getPosition();
         this.nextPostion = header.getNextPosition();
     }
 
     @Override
     public String toString() {
         StringBuilder builder = new StringBuilder();
         builder.append("{ eventId:").append(eventId);
         builder.append(",databaseName:").append(databaseName);
         builder.append(",tableName:").append(tableName);
         builder.append(",eventType:").append(eventType);
         builder.append(",timestamp:").append(timestamp);
         builder.append(",timestampReceipt:").append(timestampReceipt);
         builder.append(",position:").append(position);
         builder.append(",nextPostion:").append(nextPostion);
         builder.append(",serverId:").append(serverId);
         builder.append(",isDdl:").append(isDdl);
         builder.append(",sql:").append(sql);
         builder.append(",before:").append(before);
         builder.append(",after:").append(after).append("}");
 
         return builder.toString();
     }
 
     public long getEventId() {
         return eventId;
     }
 
     public void setEventId(long eventId) {
         this.eventId = eventId;
     }
 
     public String getDatabaseName() {
         return databaseName;
     }
 
     public void setDatabaseName(String databaseName) {
         this.databaseName = databaseName;
     }
 
     public String getTableName() {
         return tableName;
     }
 
     public void setTableName(String tableName) {
         this.tableName = tableName;
     }
 
     public int getEventType() {
         return eventType;
     }
 
     public void setEventType(int eventType) {
         this.eventType = eventType;
     }
 
     public long getTimestamp() {
         return timestamp;
     }
 
     public void setTimestamp(long timestamp) {
         this.timestamp = timestamp;
     }
 
     public long getTimestampReceipt() {
         return timestampReceipt;
     }
 
     public void setTimestampReceipt(long timestampReceipt) {
         this.timestampReceipt = timestampReceipt;
     }
 
     public long getPosition() {
         return position;
     }
 
     public void setPosition(long position) {
         this.position = position;
     }
 
     public long getNextPostion() {
         return nextPostion;
     }
 
     public void setNextPostion(long nextPostion) {
         this.nextPostion = nextPostion;
     }
 
     public long getServerId() {
         return serverId;
     }
 
     public void setServerId(long serverId) {
         this.serverId = serverId;
     }
 
     public Map<String, String> getBefore() {
         return before;
     }
 
     public void setBefore(Map<String, String> before) {
         this.before = before;
     }
 
     public Map<String, String> getAfter() {
         return after;
     }
 
     public void setAfter(Map<String, String> after) {
         this.after = after;
     }
 
     public Boolean getDdl() {
         return isDdl;
     }
 
     public void setDdl(Boolean ddl) {
         isDdl = ddl;
     }
 
     public String getSql() {
         return sql;
     }
 
     public void setSql(String sql) {
         this.sql = sql;
     }
 
 
 }

public class CDCEventManager {
 
     public static final ConcurrentLinkedDeque<LogEvent> queue = new ConcurrentLinkedDeque<LogEvent>();
 
 }

MysqlConnection：

public class MysqlConnection {
 
     private static final Logger logger = LoggerFactory.getLogger(MysqlConnection.class);
 
     private static Connection conn;
 
     private static String host;
     private static int port;
     private static String user;
     private static String password;
 
     public static void setConnection(String hostArg, int portArg, String userArg, String passwordArg) {
         try {
             if (conn == null || conn.isClosed()) {
                 Class.forName("com.mysql.jdbc.Driver");
 
                 host = hostArg;
                 port = portArg;
                 user = userArg;
                 password = passwordArg;
 
                 conn = DriverManager.getConnection("jdbc:mysql://" + host + ":" + port + "/", user, password);
                 logger.info("connected to mysql:{} : {}", user, password);
             }
         } catch (ClassNotFoundException e) {
             logger.error(e.getMessage(), e);
         } catch (SQLException e) {
             logger.error(e.getMessage(), e);
         }
     }
 
     public static Connection getConnection() {
         try {
             if (conn == null || conn.isClosed()) {
                 setConnection(host, port, user, password);
             }
         } catch (SQLException e) {
             logger.error(e.getMessage(), e);
         }
         return conn;
     }
 
     /**
      * 获取Column信息
      *
      * @return
      */
     public static Map<String, List<ColumnInfo>> getColumns() {
         Map<String, List<ColumnInfo>> cols = new HashMap<>();
         Connection conn = getConnection();
 
         try {
             DatabaseMetaData metaData = conn.getMetaData();
             ResultSet r = metaData.getCatalogs();
             String tableType[] = {"TABLE"};
             while (r.next()) {
                 String databaseName = r.getString("TABLE_CAT");
                 ResultSet result = metaData.getTables(databaseName, null, null, tableType);
                 while (result.next()) {
                     String tableName = result.getString("TABLE_NAME");
                     //                  System.out.println(result.getInt("TABLE_ID"));
                     String key = databaseName + "." + tableName;
                     ResultSet colSet = metaData.getColumns(databaseName, null, tableName, null);
                     cols.put(key, new ArrayList<ColumnInfo>());
                     while (colSet.next()) {
                         ColumnInfo columnInfo = new ColumnInfo(colSet.getString("COLUMN_NAME"),
                             colSet.getString("TYPE_NAME"));
                         cols.get(key).add(columnInfo);
                     }
 
                 }
             }
         } catch (SQLException e) {
             logger.error(e.getMessage(), e);
         }
         return cols;
     }
 
     /**
      * 参考 mysql> show binary logs +------------------+-----------+ | Log_name         | File_size |
      * +------------------+-----------+ | mysql-bin.000001 |       126 | | mysql-bin.000002 |       126 | |
      * mysql-bin.000003 |      6819 | | mysql-bin.000004 |      1868 | +------------------+-----------+
      */
     public static List<BinlogInfo> getBinlogInfo() {
         List<BinlogInfo> binlogList = new ArrayList<>();
 
         Connection conn = null;
         Statement statement = null;
         ResultSet resultSet = null;
 
         try {
             conn = getConnection();
             statement = conn.createStatement();
             resultSet = statement.executeQuery("show binary logs");
             while (resultSet.next()) {
                 BinlogInfo binlogInfo = new BinlogInfo(resultSet.getString("Log_name"), resultSet.getLong("File_size"));
                 binlogList.add(binlogInfo);
             }
         } catch (Exception e) {
             logger.error(e.getMessage(), e);
         } finally {
             try {
                 if (resultSet != null) { resultSet.close(); }
                 if (statement != null) { statement.close(); }
                 if (conn != null) { conn.close(); }
             } catch (SQLException e) {
                 logger.error(e.getMessage(), e);
             }
         }
 
         return binlogList;
     }
 
     /**
      * 参考： mysql> show master status; +------------------+----------+--------------+------------------+ | File |
      * Position | Binlog_Do_DB | Binlog_Ignore_DB | +------------------+----------+--------------+------------------+ |
      * mysql-bin.000004 |     1868 |              |                  |
      * +------------------+----------+--------------+------------------+
      *
      * @return
      */
     public static BinlogMasterStatus getBinlogMasterStatus() {
         BinlogMasterStatus binlogMasterStatus = new BinlogMasterStatus();
         Connection conn = null;
         Statement statement = null;
         ResultSet resultSet = null;
 
         try {
             conn = getConnection();
             statement = conn.createStatement();
             resultSet = statement.executeQuery("show master status");
             while (resultSet.next()) {
                 binlogMasterStatus.setBinlogName(resultSet.getString("File"));
                 binlogMasterStatus.setPosition(resultSet.getLong("Position"));
             }
         } catch (Exception e) {
             logger.error(e.getMessage(), e);
         } finally {
             try {
                 if (resultSet != null) { resultSet.close(); }
                 if (statement != null) { statement.close(); }
                 if (conn != null) { conn.close(); }
             } catch (SQLException e) {
                 logger.error(e.getMessage(), e);
             }
         }
 
         return binlogMasterStatus;
     }
 
     /**
      * 获取open-replicator所连接的mysql服务器的serverid信息
      *
      * @return
      */
     public static int getServerId() {
         int serverId = 6789;
         Connection conn = null;
         Statement statement = null;
         ResultSet resultSet = null;
         try {
             conn = getConnection();
             statement = conn.createStatement();
             resultSet = statement.executeQuery("show variables like 'server_id'");
             while (resultSet.next()) {
                 serverId = resultSet.getInt("Value");
             }
         } catch (Exception e) {
             logger.error(e.getMessage(), e);
         } finally {
             try {
                 if (resultSet != null) { resultSet.close(); }
                 if (statement != null) { statement.close(); }
                 if (conn != null) { conn.close(); }
             } catch (SQLException e) {
                 logger.error(e.getMessage(), e);
             }
         }
         return serverId;
     }
 
 }

NotificationListener:

public class NotificationListener  implements BinlogEventListener {
 
     private static Logger logger = LoggerFactory.getLogger(NotificationListener.class);
 
     @Override
     public void onEvents(BinlogEventV4 event) {
         if (event == null) {
             logger.error("binlog event is null");
             return;
         }
         int eventType = event.getHeader().getEventType();
         System.out.println("eventType---->"+ MySqlEventTypeIdToString.getInstance().get(eventType));
         switch (eventType) {
             case MySQLConstants.FORMAT_DESCRIPTION_EVENT: {
                 logger.trace("FORMAT_DESCRIPTION_EVENT");
                 break;
             }
             case MySQLConstants.TABLE_MAP_EVENT:
                 //每次ROW_EVENT前都伴随一个TABLE_MAP_EVENT事件，保存一些表信息，如tableId, tableName, databaseName, 而ROW_EVENT只有tableId
             {
                 TableMapEvent tme = (TableMapEvent)event;
                 TableInfoKeeper.saveTableIdMap(tme);
                 logger.trace("TABLE_MAP_EVENT:tableId:{}", tme.getTableId());
                 break;
             }
             case MySQLConstants.DELETE_ROWS_EVENT: {
                 DeleteRowsEvent dre = (DeleteRowsEvent)event;
                 long tableId = dre.getTableId();
                 logger.trace("DELETE_ROW_EVENT:tableId:{}", tableId);
 
                 TableInfo tableInfo = TableInfoKeeper.getTableInfo(tableId);
                 String databaseName = tableInfo.getDatabaseName();
                 String tableName = tableInfo.getTableName();
 
                 List<Row> rows = dre.getRows();
                 for (Row row : rows) {
                     List<Column> before = row.getColumns();
                     Map<String, String> beforeMap = getMap(before, databaseName, tableName);
                     if (beforeMap != null && beforeMap.size() > 0) {
                         LogEvent cdcEvent = new LogEvent(dre, databaseName, tableName);
                         cdcEvent.setDdl(false);
                         cdcEvent.setSql(null);
                         cdcEvent.setBefore(beforeMap);
                         CDCEventManager.queue.addLast(cdcEvent);
                         logger.info("cdcEvent:{}", cdcEvent);
                     }
                 }
                 break;
 
             }
             case MySQLConstants.UPDATE_ROWS_EVENT:
             {
                 UpdateRowsEvent upe = (UpdateRowsEvent)event;
                 long tableId = upe.getTableId();
                 logger.info("UPDATE_ROWS_EVENT:tableId:{}",tableId);
 
                 TableInfo tableInfo = TableInfoKeeper.getTableInfo(tableId);
                 String databaseName = tableInfo.getDatabaseName();
                 String tableName = tableInfo.getTableName();
 
                 List<Pair<Row>> rows = upe.getRows();
                 for(Pair<Row> p:rows){
                     List<Column> colsBefore = p.getBefore().getColumns();
                     List<Column> colsAfter = p.getAfter().getColumns();
                     Map<String,String> beforeMap = getMap(colsBefore,databaseName,tableName);
                     Map<String,String> afterMap = getMap(colsAfter,databaseName,tableName);
                     if(beforeMap!=null && afterMap!=null && beforeMap.size()>0 && afterMap.size()>0){
                         LogEvent cdcEvent = new LogEvent(upe,databaseName,tableName);
                         cdcEvent.setDdl(false);
                         cdcEvent.setSql(null);
                         cdcEvent.setBefore(beforeMap);
                         cdcEvent.setAfter(afterMap);
                         CDCEventManager.queue.addLast(cdcEvent);
                         logger.info("cdcEvent:{}",cdcEvent);
                     }
                 }
                 break;
             }
             case MySQLConstants.UPDATE_ROWS_EVENT_V2:
             {
                 UpdateRowsEventV2 upe = (UpdateRowsEventV2)event;
                 long tableId = upe.getTableId();
                 logger.info("UPDATE_ROWS_EVENT_V2:tableId:{}",tableId);
 
                 TableInfo tableInfo = TableInfoKeeper.getTableInfo(tableId);
                 String databaseName = tableInfo.getDatabaseName();
                 String tableName = tableInfo.getTableName();
 
                 List<Pair<Row>> rows = upe.getRows();
                 for(Pair<Row> p:rows){
                     List<Column> colsBefore = p.getBefore().getColumns();
                     List<Column> colsAfter = p.getAfter().getColumns();
                     Map<String,String> beforeMap = getMap(colsBefore,databaseName,tableName);
                     Map<String,String> afterMap = getMap(colsAfter,databaseName,tableName);
                     if(beforeMap!=null && afterMap!=null && beforeMap.size()>0 && afterMap.size()>0){
                         LogEvent cdcEvent = new LogEvent(upe,databaseName,tableName);
                         cdcEvent.setDdl(false);
                         cdcEvent.setSql(null);
                         cdcEvent.setBefore(beforeMap);
                         cdcEvent.setAfter(afterMap);
                         CDCEventManager.queue.addLast(cdcEvent);
                         logger.info("cdcEvent:{}",cdcEvent);
                     }
                 }
                 break;
             }
             case MySQLConstants.WRITE_ROWS_EVENT :
                 {
                     WriteRowsEventV2 wre = (WriteRowsEventV2)event;
                 long tableId = wre.getTableId();
                 logger.trace("WRITE_ROWS_EVENT:tableId:{}",tableId);
 
                 TableInfo tableInfo = TableInfoKeeper.getTableInfo(tableId);
                 String databaseName = tableInfo.getDatabaseName();
                 String tableName = tableInfo.getTableName();
 
                 List<Row> rows = wre.getRows();
                 for(Row row: rows){
                     List<Column> after = row.getColumns();
                     Map<String,String> afterMap = getMap(after,databaseName,tableName);
                     if(afterMap!=null && afterMap.size()>0){
                         LogEvent cdcEvent = new LogEvent(wre,databaseName,tableName);
                         cdcEvent.setDdl(false);
                         cdcEvent.setSql(null);
                         cdcEvent.setAfter(afterMap);
                         CDCEventManager.queue.addLast(cdcEvent);
                         logger.info("cdcEvent:{}",cdcEvent);
                     }
                 }
                 break;
             }
             case   MySQLConstants.WRITE_ROWS_EVENT_V2:
             {
                 WriteRowsEventV2 wre = (WriteRowsEventV2)event;
                 long tableId = wre.getTableId();
                 logger.trace("WRITE_ROWS_EVENT:tableId:{}",tableId);
 
                 TableInfo tableInfo = TableInfoKeeper.getTableInfo(tableId);
                 String databaseName = tableInfo.getDatabaseName();
                 String tableName = tableInfo.getTableName();
 
                 List<Row> rows = wre.getRows();
                 for(Row row: rows){
                     List<Column> after = row.getColumns();
                     Map<String,String> afterMap = getMap(after,databaseName,tableName);
                     if(afterMap!=null && afterMap.size()>0){
                         LogEvent cdcEvent = new LogEvent(wre,databaseName,tableName);
                         cdcEvent.setDdl(false);
                         cdcEvent.setSql(null);
                         cdcEvent.setAfter(afterMap);
                         CDCEventManager.queue.addLast(cdcEvent);
                         logger.info("cdcEvent:{}",cdcEvent);
                     }
                 }
                 break;
             }
             case MySQLConstants.QUERY_EVENT:
             {
                 QueryEvent qe = (QueryEvent)event;
                 TableInfo tableInfo = createTableInfo(qe);
                 if(tableInfo == null){
                     break;
                 }
 
                 String databaseName = tableInfo.getDatabaseName();
                 String tableName = tableInfo.getTableName();
                 logger.trace("QUERY_EVENT:databaseName:{},tableName:{}",databaseName,tableName);
 
                 LogEvent cdcEvent = new LogEvent(qe,databaseName,tableName);
                 cdcEvent.setDdl(true);
                 cdcEvent.setSql(qe.getSql().toString());
                 CDCEventManager.queue.addLast(cdcEvent);
                 logger.info("cdcEvent:{}",cdcEvent);
                 break;
             }
             case MySQLConstants.XID_EVENT:{
                 XidEvent xe = (XidEvent)event;
                 logger.trace("XID_EVENT: xid:{}",xe.getXid());
                 break;
             }
             default:
             {
                 logger.trace("DEFAULT:{}",eventType);
                 break;
             }
 
         }
     }
 
 
     /**
      * ROW_EVENT中是没有Column信息的，需要通过MysqlConnection（下面会讲到）的方式读取列名信息，
      * 然后跟取回的List<Column>进行映射。
      *
      * @param cols
      * @param databaseName
      * @param tableName
      * @return
      */
     private Map<String,String> getMap(List<Column> cols, String databaseName, String tableName){
         Map<String,String> map = new HashMap<>();
         if(cols == null || cols.size()==0){
             return null;
         }
         String fullName = databaseName+"."+tableName;
         List<ColumnInfo> columnInfoList = TableInfoKeeper.getColumns(fullName);
         if(columnInfoList == null)
             return null;
         if(columnInfoList.size() != cols.size()){
             TableInfoKeeper.refreshColumnsMap();
             if(columnInfoList.size() != cols.size())
             {
                 logger.warn("columnInfoList.size is not equal to cols.");
                 return null;
             }
         }
 
         for(int i=0;i<columnInfoList.size(); i++){
             if(cols.get(i).getValue()==null)
                 map.put(columnInfoList.get(i).getName(),"");
             else
                 map.put(columnInfoList.get(i).getName(), cols.get(i).toString());
         }
 
         return map;
     }
 
     /**
      * 从sql中提取Table信息，因为QUERY_EVENT是对应DATABASE这一级别的，不像ROW_EVENT是对应TABLE这一级别的，
      * 所以需要通过从sql中提取TABLE信息,封装到TableInfo对象中
      *
      * @param qe
      * @return
      */
     private TableInfo createTableInfo(QueryEvent qe){
         String sql = qe.getSql().toString().toLowerCase();
         if("begin".equals(sql)){
             return null;
         }
         TableInfo ti = new TableInfo();
         String databaseName = qe.getDatabaseName().toString();
         String tableName = null;
         if(checkFlag(sql,"table")){
             tableName = getTableName(sql,"table");
         } else if(checkFlag(sql,"truncate")){
             tableName = getTableName(sql,"truncate");
         } else{
             logger.warn("can not find table name from sql:{}",sql);
             return null;
         }
         ti.setDatabaseName(databaseName);
         ti.setTableName(tableName);
         ti.setFullName(databaseName+"."+tableName);
 
         return ti;
     }
 
     private boolean checkFlag(String sql, String flag){
         String[] ss = sql.split(" ");
         for(String s:ss){
             if(s.equals(flag)){
                 return true;
             }
         }
         return false;
     }
 
     private String getTableName(String sql, String flag){
         String[] ss = sql.split("\\.");
         String tName = null;
         if (ss.length > 1) {
             String[] strs = ss[1].split(" ");
             tName = strs[0];
         } else {
             String[] strs = sql.split(" ");
             boolean start = false;
             for (String s : strs) {
                 if (s.indexOf(flag) >= 0) {
                     start = true;
                     continue;
                 }
                 if (start && !s.isEmpty()) {
                     tName = s;
                     break;
                 }
             }
         }
         tName.replaceAll("`", "").replaceAll(";", "");
 
         //del "("[create table person(....]
         int index = tName.indexOf('(');
         if(index>0){
             tName = tName.substring(0, index);
         }
 
         return tName;
     }
 
 }

TableInfo：

public class TableInfo {
 
     private String databaseName;
     private String tableName;
     private String fullName;
     // 省略Getter和Setter
 
     public String getDatabaseName() {
         return databaseName;
     }
 
     public void setDatabaseName(String databaseName) {
         this.databaseName = databaseName;
     }
 
     public String getTableName() {
         return tableName;
     }
 
     public void setTableName(String tableName) {
         this.tableName = tableName;
     }
 
     public String getFullName() {
         return fullName;
     }
 
     public void setFullName(String fullName) {
         this.fullName = fullName;
     }
 
     @Override
     public boolean equals(Object o) {
         if (this == o) { return true; }
         if (o == null || getClass() != o.getClass()) { return false; }
         TableInfo tableInfo = (TableInfo)o;
         return Objects.equals(databaseName, tableInfo.databaseName) &&
             Objects.equals(tableName, tableInfo.tableName) &&
             Objects.equals(fullName, tableInfo.fullName);
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(databaseName, tableName, fullName);
     }
 }

TableInfoKeeper:

public class TableInfoKeeper {
 
     private static final Logger logger = LoggerFactory.getLogger(TableInfoKeeper.class);
 
     private static Map<Long, TableInfo> tabledIdMap = new ConcurrentHashMap<>();
     private static Map<String, List<ColumnInfo>> columnsMap = new ConcurrentHashMap<>();
 
     static {
         columnsMap = MysqlConnection.getColumns();
     }
 
     public static void saveTableIdMap(TableMapEvent tme) {
         long tableId = tme.getTableId();
         tabledIdMap.remove(tableId);
 
         TableInfo table = new TableInfo();
         table.setDatabaseName(tme.getDatabaseName().toString());
         table.setTableName(tme.getTableName().toString());
         table.setFullName(tme.getDatabaseName() + "." + tme.getTableName());
 
         tabledIdMap.put(tableId, table);
     }
 
     public static synchronized void refreshColumnsMap() {
         Map<String, List<ColumnInfo>> map = MysqlConnection.getColumns();
         if (map.size() > 0) {
             //          logger.warn("refresh and clear cols.");
             columnsMap = map;
             //          logger.warn("refresh and switch cols:{}",map);
         } else {
             logger.error("refresh columnsMap error.");
         }
     }
 
     public static TableInfo getTableInfo(long tableId) {
         return tabledIdMap.get(tableId);
     }
 
     public static List<ColumnInfo> getColumns(String fullName) {
         return columnsMap.get(fullName);
     }
 
 }