cnblogs.com/hebaibai/p/10911899.html

  业务背景

  写任何工具都不能脱离实际业务的背景。开始这个项目的时候是因为现有的项目中数据分布太零碎，零零散散的分布在好几个数据库中，没有统一的数据库来收集这些数据。这种情况下想做一个大而全的会员中心系统比较困难。（这边是一个以互联网保险为中心的项目，保单，会员等数据很零散的储存在好几个项目之中，并且项目之间的数据基本上是隔离的）。

  现有的项目数据库是在腾讯云中储存，虽然腾讯提供了数据同步功能，但是这样必须要表结构相同才行，并不符合我们的需求。所以需要自行开发。

  需求技术选择

  这个任务交给了我和另外一个同事来做。

  同事的

  同事希望可以通过ETL工具Kettle来做，这个东西我没有研究过，是同事自己在研究。具体过程不是很清楚，但是最后是通过在mysql中设置更新，修改，删除的触发器，然后在Kettle中做了一个定时任务，实现了数据同步的功能，初步测试符合需求。但是必须要在数据库中设置触发器，并且会有一个临时表，这一点我个人不是很喜欢。

  我的

  我是本着能自己写就自己写的原则，准备自己写一个。刚开始使用的是定时任务比较两个库的数据差别，然后再同步数据。但是经过一定的数据测试后，发现在数据量大的时候，定时任务中的上一个任务没有执行完毕，下一个任务就又开始了。这样造成了两边数据不一致。最终这个方案废弃了。

  后来通过研究，发现mysql的数据操作会记录在binlog中，这时就有了新的方案。可以通过逐行获取binlog信息，经过解析数据后，同步在目标库中。

  既然有了方案，那么就开始做吧。

  开始尝试1

  首先要打开数据库的binlog功能，这一步比较简单，修改mysql的配置文件：/etc/mysql/mysql.conf.d/mysqld.cnf，添加：

server-id        = 1<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />log_bin            = /var/log/mysql/mysql-bin.log<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />expire_logs_days    = 10<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />max_binlog_size         = 100M<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />binlog_format           = ROW<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />

  然后重启mysql 就好了，具体每个参数的意思，搜索一下就好了。这时候随意的对某一个数据库中的表做一下增删改，对应的日志就会记录在/var/log/mysql/这个文件夹下了。我们看一下这个文件夹里的东西：

  这里的文件是没有办法正常查看的，需要使用mysql提供的命令来查看，命令是这个样子的：

  1、查看

mysqlbinlog mysql-bin.000002<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />

  2、指定位置查看

mysqlbinlog --start-position="120" --stop-position="332" mysql-bin.000002<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />

  因为我们现在的binlog_format指定的格式是ROW（就在上面写的，还记得吗？），所谓binlog文件的内容没有办法正常查看，因为他是这个样子的：

  这时，我们需要对输出进行解码

mysqlbinlog --base64-output=decode-rows -v mysql-bin.000001<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />

  这时候，显示的结果就变成了：

  虽然还不是正常的sql，但是好赖是有一定的格式了。

  but自己来做解析的话还是很麻烦，so～放弃这种操作。

  继续尝试2

  经过再次研究后，发现数据库中执行sql也是可以查看binlog的。主要有如下几条命令：

--重置binlog<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />reset master;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />--查看binlog的配置<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />show variables like '%binlog%';<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />--查看所有的binlog<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />show binary logs;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />--查看正在写入的binlog<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />show master status;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />--查看指定binlog文件<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />show binlog events in 'mysql-bin.000001';<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />--查看指定binlog文件，并指定位置<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />show binlog events in 'mysql-bin.000001' from [pos] limit [显示多少条];<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />

  按照上面的命令执行结果为：

  发现sql还是不能正常显示。这里的原因应该是binlog_format配置的原因。将其修改为 binlog_format=Mixed后，完美解决。经过数据库中一通增删改后，显示的sql类似这样：

use pay; /* ApplicationName=DataGrip 2018.2.5 */ UPDATE pay.p_pay_log t SET t.mark_0 = 'sdfsdf' WHERE t.id LIKE '342' ESCAPE '#'<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />

  现在似乎已经可以开始写数据同步了，只要在启动的时候获取当正在使用的是哪一个日志文件，记录binlog的位置，然后一点一点向下执行，解析sql就好了。但是在这个过程中，我发现阿里巴巴有一款开源的软件可以用。就是标题上说道的：canal。看了一下网站上的介绍，简直美滋滋。

  它的文档和代码地址在这里：，大家可以看一下。现在就准备用这个来完成我所需要的功能。

  正式开始写

  首先看一下介绍，canal是需要单独运行一个服务的，这个服务具体的配置还是比较简单的。它的作用我自己理解就是监控binlog，然后根据自己的需要获取binlog中一定量的数据。这个数据是经过处理的，可以比较方便的知道里面的具体信息。比如那些数据发生了变动，每列数据的列名是什么，变动前和变动后的值是啥之类的。那么开始。

  1.我的想法

  1）项目启动的时候，开启canal的链接，以及初始化一些配置。

@Bean<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />public CanalConnector canalConnector() {<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    CanalConnector connector = CanalConnectors.newSingleConnector(<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            //对应canal服务的链接<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            new InetSocketAddress(canalConf.getIp(), canalConf.getPort()),<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            //链接的目标，这里对应canal服务中的配置，需要查阅文档<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            canalConf.getDestination(),<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            //不知道是什么用户，使用“”<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            canalConf.getUser(),<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            //不知道是什么密码，使用“”<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            canalConf.getPassword()<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    );<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    return connector;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />}<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />

  2）先开启一个线程，里面写一个死循环，用于从canal的服务中获取binlog中的消息。这个消息类是：com.alibaba.otter.canal.protocol.Message。

Message message = connector.getWithoutAck(100);<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />

  3）取出Message中的事件集合，就是binlog中的每一条数据。将类型为增删改的数据取出，之后每一条数据放在一个线程中，用线程池去执行它。

List entries = message.getEntries();<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />message.getEntries()：从链接中获取的数据集合，每一条代表1条binlog数据<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />

  4）在每一个线程中，取出Entry中的数据，根据其类型拼接各种sql，并执行。

Header header = entry.getHeader();<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />//获取发生变化的表名称，可能会没有<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />String tableName = header.getTableName();<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />//获取发生变化的数据库名称，可能会没有<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />String schemaName = header.getSchemaName();<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />//获取事件类型<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />EventType eventType = rowChange.getEventType();<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />/**<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />这里我们只是用其中的三种类型：<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    EventType.DELETE 删除<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    EventType.INSERT 插入<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    EventType.UPDATE 更新<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />*/<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />//获取发生变化的数据<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />//遍历其中的数据<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />int rowDatasCount = rowChange.getRowDatasCount();<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />for (int i = 0; i  executorService.execute(job.newTask(entry)));<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    }<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />}<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />

  这里的jobList是我自己定义List，代码如下：

package com.hebaibai.miner.job;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />import com.alibaba.fastjson.JSONObject;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />import com.alibaba.otter.canal.protocol.CanalEntry;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />import com.google.protobuf.InvalidProtocolBufferException;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />import lombok.Data;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />import lombok.extern.slf4j.Slf4j;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />import org.springframework.jdbc.core.JdbcTemplate;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />import static com.alibaba.otter.canal.protocol.CanalEntry.Entry;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />@Slf4j<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />@Data<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />public abstract class Job {<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    /**<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     * 数据库链接<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     */<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    protected JdbcTemplate jdbcTemplate;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    /**<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     * 额外配置<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     */<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    protected JSONObject prop;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    /**<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     * 校验目标是否为合适的数据库和表<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     *<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     * @param table<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     * @param database<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     * @return<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     */<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    abstract public boolean isMatches(String table, String database);<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    /**<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     * 实例化一个Runnable<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     *<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     * @param entry<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     * @return<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     */<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    abstract public Runnable newTask(final Entry entry);<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    /**<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     * 获取RowChange<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     *<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     * @param entry<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     * @return<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     */<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    protected CanalEntry.RowChange getRowChange(Entry entry) {<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        try {<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            return CanalEntry.RowChange.parseFrom(entry.getStoreValue());<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        } catch (InvalidProtocolBufferException e) {<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            e.printStackTrace();<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        }<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        return null;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    }<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />}<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />

  jobList里面放的是Job的实现类。

  3.写一个Job的实现类，并用于同步表，并转换字段名称。

  因为需求中要求两个同步的数据中可能字段名称不一致，所以我写了一个josn用来配置两个表的字段对应关系：

//省略其他配置<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />"prop": {<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />//来源数据库<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />  "database": "pay",<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />//来源表<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />  "table": "p_pay_msg",<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />//目标表（目标库在其他地方配置）<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />  "target": "member",<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />//字段对应关系<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />//key  ：来源表的字段名<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />//value：目标表的字段名<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />  "mapping": {<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    "id": "id",<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    "mch_code": "mCode",<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    "send_type": "mName",<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    "order_id": "phone",<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    "created_time": "create_time",<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    "creator": "remark"<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />  }<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />}<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />//省略其他配置<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />

  下面是全部的代码，主要做的就是取出变动的数据，按照对应的字段名重新拼装sql，然后执行就好了，不多解释。扩展：

package com.hebaibai.miner.job;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />import com.alibaba.fastjson.JSONObject;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />import lombok.extern.slf4j.Slf4j;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />import org.apache.commons.lang3.StringUtils;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />import java.util.ArrayList;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />import java.util.List;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />import java.util.stream.Collectors;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />import static com.alibaba.otter.canal.protocol.CanalEntry.*;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />/**<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /> * 单表同步，表的字段名称可以不同，类型需要一致<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /> * 表中需要有id字段<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /> */<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />@SuppressWarnings("ALL")<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />@Slf4j<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />public class TableSyncJob extends Job {<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    /**<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     * 用于校验是否适用于当前的配置<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     *<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     * @param table<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     * @param database<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     * @return<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     */<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    @Override<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    public boolean isMatches(String table, String database) {<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        return prop.getString("database").equals(database) &&<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />                prop.getString("table").equals(table);<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    }<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  /><br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    /**<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     * 返回一个新的Runnable<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     *<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     * @param entry<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     * @return<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />     */<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    @Override<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    public Runnable newTask(final Entry entry) {<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        return () -> {<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            RowChange rowChange = super.getRowChange(entry);<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            if (rowChange == null) {<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />                return;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            }<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            EventType eventType = rowChange.getEventType();<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            int rowDatasCount = rowChange.getRowDatasCount();<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            for (int i = 0; i  column.getUpdated() || column.getIsKey()).collect(Collectors.toList());<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        //根据表映射关系拼装更新sql<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        JSONObject mapping = prop.getJSONObject("mapping");<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        String target = prop.getString("target");<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        List columnNames = new ArrayList();<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        List columnValues = new ArrayList();<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        for (int i = 0; i  column.getUpdated()).collect(Collectors.toList());<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        //找到之前的数据中的keys<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        List keyCols = before.stream().filter(column -> column.getIsKey()).collect(Collectors.toList());<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        //没有key,执行更新替换<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        if (keyCols.size() == 0) {<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            return;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        }<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        //根据表映射关系拼装更新sql<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        JSONObject mapping = prop.getJSONObject("mapping");<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        String target = prop.getString("target");<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        //待更新数据<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        List updatas = new ArrayList();<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        for (int i = 0; i  column.getIsKey()).collect(Collectors.toList());<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        if (keyCols.size() == 0) {<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            return;<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        }<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        //根据表映射关系拼装更新sql<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        JSONObject mapping = prop.getJSONObject("mapping");<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        String target = prop.getString("target");<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        StringBuilder sql = new StringBuilder();<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        sql.append("DELETE FROM ").append(target).append(" WHERE ");<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        List where = new ArrayList();<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        for (Column column : keyCols) {<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            String name = mapping.getString(column.getName());<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />            where.add(name + " = '" + column.getValue() + "' ");<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        }<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        sql.append(StringUtils.join(where, "and "));<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        String sqlStr = sql.toString();<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        log.debug(sqlStr);<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />        jdbcTemplate.execute(sqlStr);<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />    }<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />}<br style="font-size: inherit;color: inherit;line-height: inherit;overflow-wrap: inherit !important;word-break: inherit !important;"  />

  项目源码

  END

  Java面试题专栏

我知道你 “在看”