干货！canal-adapter 1.1.4 应用存在的缺陷及问题修复

前言

相信点击进来看这边文章的同学们，目的是很明确的，并且已经部署canal-deploy或者canal-server，并且成功消费了MySQL的binlog日志，通过tcp方式暴露或者投递到mq了，当然，建议这块引入canal-admin，真的很香~

问题的发现

  • 同步rdb，采用的非镜像模式，不会同步ddl语句
    这个问题相信是很容易被发现的，一般情况下，我们不会采用canal做整库的镜像同步，我们更多时候使用canal是下图的场景


    图1.png

    可以发现，我们是把不同业务库的表同步至一个聚合库中，这种模式便不能使用canal的镜像模式，然鹅，只有镜像模式才会同步ddl，贴上源码


    图2.png

    可以看到，当判断是ddl语句时，只是清除了目标库的表结构缓存。
    这个时候就出现问题了，当我们没有及时同步业务聚合库的表结构，此时新的dml语句过来，在拼接sql语句的时候由于在遍历源表字段列表的时候需要取目标表字段类型的值不存在，于是抛出一个异常
 for (Map.Entry<String, String> entry : columnsMap.entrySet()) {
            String targetColumnName = entry.getKey();
            String srcColumnName = entry.getValue();
            if (srcColumnName == null) {
                srcColumnName = Util.cleanColumn(targetColumnName);
            }

            Integer type = ctype.get(Util.cleanColumn(targetColumnName).toLowerCase());
            //此处匹配不到则抛出异常
            if (type == null) {
                throw new RuntimeException("Target column: " + targetColumnName + " not matched");
            }
            Object value = data.get(srcColumnName);
            BatchExecutor.setValue(values, type, value);
        }

这里我们不对这种处理方式的好坏做评价，我们思考下，要如何避免异常的抛出呢，有如下两种解决方式：

  1. 实现非镜像模式下ddl的同步
    我在上面ddl语句块内，在删除缓存前，将alter打头的ddl在目标表进行同步，如下


    C37CB1B2-F05A-44f6-AB0E-782406035133.png
    6E48B33D-DB5B-4764-9BF8-67263517996D.png
  2. 在拼接sql语句时，只拼接目标表存在的字段
    通过上述第一种方法确实已经可以解决表结构不一致带来的同步失败，然鹅，我也在考虑为什么canal开发人员不这么做呢，我认为有几个原因：1）一般不建议将ddl操作让程序去跑，但这条我认为也不是必须严格执行；2）ddl操作会造成临时的锁表，如果是线上运行，并且由程序控制，显得有点危险
    于是，如果有同学认为上述方式不太合适，也可以选择下面这种方式，那就是将抛出异常改为如下
  if (type == null) {
      continue;
  }

没错，就是不拼接你这个字段就完了，我也搞不懂，为啥canal作者不这么做，感觉这样子挺友好的，为啥抛出异常让整个同步失效，因为一般情况下，如果你想要同步这个字段，那你在后续代码编写的时候自然就会发现并且弥补上，那如果这个字段你不需要同步，那就放那边呗，等以后你发现需要用到的时候再处理也行，挺友好的

小结

其实如果有去阅读源码的话，从上面rdb类型的同步中，大家可以发现，canal所谓的实现数据的同步并没有用到多高深的技巧，我们可以发现adapter的application.yml配置文件中，将源库跟目标库都配置上了，目的就是在同步的时候还需要查一下源库，校验的同时还顺便取了下数据，这就感觉，其实canal做的比较重要的就是通过row模式的binlog读取，告知各位订阅者，某条数据发生了数据的变更，然后各个下游来通过变更的类型来源库取一下数据，做下校验，再同步变更，因为既然配置了源库，那我其实很多东西都可以直接在源库操作了，你只要给我个id跟执行类型就行了，确实有这种感觉，当然，这个只是吐槽，其实canal最核心的内容是在canal server那块解析binlog的部分

  • 同步es，sql语句中select的主键字段为非简单字段删除同步失败
    该问题的发现是由于我工作中，出现了一种场景，需要将不同业务表聚合到一个ES的索引中，这个索引有点类似于HBASE的宽表，由于我们一般采用的是将库中的id字段配置为索引文档主键的方式，这就会出现不同的业务表存在相同id字段，后续同步过来的数据会覆盖相同id的数据
    其实我们思考下，这个本身就是个伪命题，一般情况我们在设计表结构的时候如果是业务主体信息表，都需要有个业务code字段来作为真正的主键字段，那一般情况下虽然是不同的表，这个code理论上冲突的概率其实非常的小。但是好死不死，我们早期设计表结构就是缺了这个字段，于是，就面临这个问题了，我需要用concat函数将id+type拼接起来作为文档的主键，这个其实是canal支持语法配置的


    C0D0E286-CC33-44cf-93DB-A292A654641D.png

    然鹅，这个配置会导致物理删除记录不能同步的！！
    那为啥这样子ES就不能同步呢，其实很容易理解，例如你删除了个商品数据，他的拼接后的_id是123_1，这个时候大家要知道，es同步配置的sql是在adapter消费端的，当我读取kafka的message并且解析后是能获取到被删除数据的基本信息（id=123，type=1），但由于用了concat函数，_id的值目前还是得不到的，如果是更新操作，通过配置的sql加上主键的限定是可以获取到这条select出来主键是123_1的数据的，但是删除的话就没了啊，就导致删除不能同步成功。
    解决：大家应该很容易想出解决方式，并且是个有局限性的方式，就是直接基于concat函数的应用来单独处理，将sql的concat拼接实现在Java代码层，一种挺暴力直接的方式，在mainTableDelete方法上复制一个自定义的方法mainTableDeletecustomize，并且_id直接硬拼出来，上code


    809206B8-E3EF-4860-AD3F-79AA6E0A4E9D.png
private String getIdSqlValFromExpr(ESMapping esMapping, Dml dml) {
        try {
            Object o = dml.getData().get(0).get(dml.getPkNames().get(0).toString());
            if(Objects.isNull(o)){
                throw new RuntimeException("无法获取kafka中删除行主键数据");
            }
            String idColumn = esMapping.get_id()!=null?esMapping.get_id():esMapping.getPk()!=null?esMapping.getPk():"_id";
            //mysql字段为ES主键
            FieldItem fieldItem = esMapping.getSchemaItem().getSelectFields().get(idColumn);
            if(!fieldItem.getExpr().contains("CONCAT")){
                //用concat函数拼接主键，此处如果不满足普通的concat句式则跑出异常
 throw new RuntimeException("函数作用主键仅支持concat函数方式");
            }
            String str = fieldItem.getExpr();
            String quStr=str.substring(str.indexOf("(")+1,str.indexOf(")"));
            String[] split = quStr.split(",");
            for(int i =0;i<split.length;i++){
                if(!split[i].contains("'")){
                    //不包含''静态字符，则认为是字段名称，进行替换
                    split[i] = o.toString();
                }else{
                    split[i] = split[i].replace("'","").trim();
                }

            }
            StringBuffer sb = new StringBuffer();
            for(int i =0;i<split.length;i++){
                sb.append(split[i]);
            }

            return sb.toString();
        }catch (Exception e){
            throw new RuntimeException("sql的函数作用主键不满足句式要求");
        }


    }

其实就是通过dml数据结合配置的sql,强行拼出es文档的_id字段值，然后再去才能正确的执行同步删除

  • 同步es，sql语句中使用group_concat组合子表数组字段数据，删除最后一条子数据，同步失败
    这个问题的发现是由于我这边需要实现一类场景，根据菜谱信息以及菜谱打标关系表，同步es的菜谱索引，并且聚合标签内容在tags字段上，canal adapter当然也是对类需求做了支持


    A9D53799-6E57-4090-867E-86DD5347265B.png

    通过group_concat聚合子表标签字段
    然鹅，通过测试发现，例如存在一个西红柿炒蛋的tags字段值[西红柿，鸡蛋]，删除了西红柿记录，此时同步是正常，当再删除掉鸡蛋记录时，同步就出现问题了失败了，只有当某条dml语句执行结束后在源库使得西红柿炒蛋菜谱是有标签值的情况下，才能又恢复正常的同步
    那为啥会这样呢，我们看一个完整的sql语句

SELECT
    CONCAT( t.product_key_word_code, '_', t.token_source ) AS _id,
    t.delete_flag AS delete_flag,
    t.token_source AS token_source,
    t.product_key_word_code AS product_key_word_code,
    t.product_key_word AS product_key_word,
    t1.category_name AS category_name 
FROM
    bas_product_key_word_info t
    LEFT JOIN 
    (
    SELECT
    product_key_word_code,
    GROUP_CONCAT( category_name ) category_name 
FROM
    bas_category_product_key_word_relation 
GROUP BY
    product_key_word_code

) t1 ON t1.product_key_word_code = t.product_key_word_code

在canal adapter中，上述的sql是会被分成一个主sql根据一个叫做subQuerySql的语句，主sql就是整个sql了，subQuerySql就是left join的子语句，由于此时收到的dml语句是对子表的删除，因此canal就拿出这个subQuerySql加上关联外键（也就是西红柿炒蛋的product_key_word_code值）的限制要查询出当前关联的标签值，然后组装下es文档更新请求塞入最新值就更新成功了！这样乍一看没问题，问题还是出在当西红柿炒蛋的所有标签数据都不存在了， 那这个单表查询就空了，而canal都是拿这个resultSet进入循环处理的，没有数据就直接跑完了，其实就是没有考虑到这种情况，因此修复如下，复制subTableSimpleFieldOperation方法，新增为subTableSimpleFieldOperationByDelete，并在delete逻辑中删除关联表数据中替换


4BF3D17F-B739-483c-BD4D-7735FE31C8C1.png

方法内针对拼接subQuerySql取数为空的情况提前处理，并复用joinTableSimpleFieldOperation方法更新为空值

 /**
     * 删除操作调用，关联子查询, 主表简单字段operation
     *
     * @param config es配置
     * @param dml dml信息
     * @param data 单行dml数据
     * @param old 单行old数据
     * @param tableItem 当前表配置
     */
    private void subTableSimpleFieldOperationByDelete(ESSyncConfig config, Dml dml, Map<String, Object> data,
                                              Map<String, Object> old, TableItem tableItem) {
        ESMapping mapping = config.getEsMapping();
        StringBuilder sql = new StringBuilder(
            "SELECT * FROM (" + tableItem.getSubQuerySql() + ") " + tableItem.getAlias() + " WHERE ");

        for (FieldItem fkFieldItem : tableItem.getRelationTableFields().keySet()) {
            String columnName = fkFieldItem.getColumn().getColumnName();
            Object value = esTemplate.getValFromData(mapping, data, fkFieldItem.getFieldName(), columnName);
            ESSyncUtil.appendCondition(sql, value, tableItem.getAlias(), columnName);
        }
        int len = sql.length();
        sql.delete(len - 5, len);
        DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
        if (logger.isTraceEnabled()) {
            logger.trace("Join table update es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
                config.getDestination(),
                dml.getTable(),
                mapping.get_index(),
                sql.toString().replace("\n", " "));
        }

        try (Connection conn = ds.getConnection();
             Statement stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)) {
            stmt.setFetchSize(Integer.MIN_VALUE);
            try (ResultSet rs = stmt.executeQuery(sql.toString())) {
                if(!rs.next()){
                    //子表无法查出数据，则将主表select 子表字段设置为null
                    Map<String, Object> esFieldData = new LinkedHashMap<>();
                    for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
                        esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), null);
                    }

                    joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData);

                }
            }
        } catch (Exception e) {
            logger.error("sqlRs has error, sql: {} ", sql);
            throw new RuntimeException(e);
        }

        Util.sqlRS(ds, sql.toString(), rs -> {
            try {
                while (rs.next()) {
                    Map<String, Object> esFieldData = new LinkedHashMap<>();

                    for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
                        if (old != null) {
                            out: for (FieldItem fieldItem1 : tableItem.getSubQueryFields()) {
                                for (ColumnItem columnItem0 : fieldItem.getColumnItems()) {
                                    if (fieldItem1.getFieldName().equals(columnItem0.getColumnName()))
                                        for (ColumnItem columnItem : fieldItem1.getColumnItems()) {
                                            if (old.containsKey(columnItem.getColumnName())) {
                                                Object val = esTemplate.getValFromRS(mapping,
                                                    rs,
                                                    fieldItem.getFieldName(),
                                                    fieldItem.getColumn().getColumnName());
                                                esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
                                                break out;
                                            }
                                        }
                                }
                            }
                        } else {
                            Object val = esTemplate.getValFromRS(mapping,
                                rs,
                                fieldItem.getFieldName(),
                                fieldItem.getColumn().getColumnName());
                            esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
                        }
                    }

                    Map<String, Object> paramsTmp = new LinkedHashMap<>();
                    for (Map.Entry<FieldItem, List<FieldItem>> entry : tableItem.getRelationTableFields().entrySet()) {
                        for (FieldItem fieldItem : entry.getValue()) {
                            if (fieldItem.getColumnItems().size() == 1) {
                                Object value = esTemplate.getValFromRS(mapping,
                                    rs,
                                    fieldItem.getFieldName(),
                                    entry.getKey().getColumn().getColumnName());
                                String fieldName = fieldItem.getFieldName();
                                // 判断是否是主键
                                if (fieldName.equals(mapping.get_id())) {
                                    fieldName = "_id";
                                }
                                paramsTmp.put(fieldName, value);
                            }
                        }
                    }

                    if (logger.isDebugEnabled()) {
                        logger.trace("Join table update es index by query sql, destination:{}, table: {}, index: {}",
                            config.getDestination(),
                            dml.getTable(),
                            mapping.get_index());
                    }
                    esTemplate.updateByQuery(config, paramsTmp, esFieldData);
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
            return 0;
        });
    }

环境准备

针对canal同步rdb或者es的部署配置各位应该都很清晰了，这也是看这篇文章的前提，那么上面针对源码的修改要如何部署呢，分为以下几个步骤

  • 环境准备
    由于GitHub的clone速度太慢了，建议通过码云协助拉取GitHub的canal项目至码云空间，再从码云clone代码


    FA75972F-E042-46cb-9191-3BB4EAEF1646.png

    canal 克隆下来后check到1.1.4稳定版，目前是基于该版本进行优化，并且从文件目录中复制canal-adapter目录至新目录并导入idea中


    0A15C115-C590-461c-BB68-C651D60A6EE5.png

    5FCDD3EC-9728-48ce-89E4-E8270D8DB0B9.png
F9404BB8-FD20-4401-9B1C-7B11086E97FE.png

再将canal-adapter 1.1.4上传至公司gitlab中迭代

构建部署

一般情况下我们只需要改动elasticsearch、hbase、rdb这三个模块，例如上述问题1是改动rdb模块，问题2、3改动elasticsearch模块，部署时，只需要在代码改完且测试通过后，对模块install，找到rdb底下的client-adapter.xxx-1.1.4-jar-with-dependencies.jar文件，然后在服务器进行替换并重启canal-adapter即可


4F964E8A-6E54-46d6-A4BF-76DDEFD76E7C.png

7CF31FF1-E810-43f0-8F97-439835C934D4.png

待解决的问题

  • canal adapter支持es同步时嵌套文档的组装，在不做父子文档双重分页的情况下，是更优的一种查询结构


    B41FEFBF-47DC-4f4b-AD5C-42E44139344B.png

然鹅，嵌套文档的同步跟上述问题3一样，存在清不干净的情况，这个解决的思路应该是一样的，有待后续处理。。。

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 212,185评论 6赞 493
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 90,445评论 3赞 385
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 157,684评论 0赞 348
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 56,564评论 1赞 284
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 65,681评论 6赞 386
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 49,874评论 1赞 290
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 39,025评论 3赞 408
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 37,761评论 0赞 268
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 44,217评论 1赞 303
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,545评论 2赞 327
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,694评论 1赞 341
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 34,351评论 4赞 332
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,988评论 3赞 315
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,778评论 0赞 21
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,007评论 1赞 266
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 46,427评论 2赞 360
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 43,580评论 2赞 349