Java 接入 canal

canal.adapter | 问题

canal.adapter | 原理
  • canal.adapter 会给它配置一个配置文件 shop.yml；
  • shop.yml 中会定义当 canal.deployer 同步过来 MySQL 的 binlog 时 canal.adapter 要执行的 SQL，这里定义的 SQL 长这样：
    select a.id,a.name,a.tags,concat(a.latitude,',',a.longitude) as location,a.remark_score,a.price_per_man,a.category_id,b.name as category_name,a.seller_id,c.remark_score as seller_remark_score,c.disabled_flag as seller_disabled_flag from shop a inner join category b on a.category_id = b.id inner join seller c on c.id = a.seller_id
canal.adapter | 问题
  • 一整条 SQL 要查出来这么多的字段，但是 canal.adapter 只查变更的字段；
  • 并且，如果 SQL 中有 a.name, b.name 的话，canal.adapter 无法区分这是两张表的两个字段，会把 a.name 和 b.name 同时更新成新的值；
canal.adapter | 问题 | 验证步骤
  • 清空 ElasticSearch 中索引 shop 的全部文档；
  • 修改 MySQL 中 shop 表的一条记录的 name 字段的值；
  • canal.adapter 索引进 ElasticSearch 的文档只有更新的 name 字段；
canal.adapter | 问题 | 解决方案
  • 不用 canal.adapter 同步数据到 ElasticSearch 中了；
  • 通过引入 canal 的依赖，在 Java 程序中，自定义的将 canal.deployer 同步过来的 binlog 索引进 ElasticSearch 中；
canal.adapter | 适用场景
  • 如果是简单的将 MySQL 中的一张表和 ElasticSearch 中的一个索引对应，用 canal.adapter 还是可以的；

SpringBoot 中引入 canal

  • 目前 mvn repository 中最新的依赖版本是 1.1.4，刚好之前自己编译的 canal 也是 1.1.4，那就正好，不用重启 canal.deployer 的其他版本；
  • 注意 canal.deployer 的版本要和引入依赖的版本一致；
canal | 依赖
<dependency>
    <groupId>com.alibaba.otter</groupId>
    <artifactId>canal.client</artifactId>
    <version>1.1.4</version>
</dependency>
<dependency>
    <groupId>com.alibaba.otter</groupId>
    <artifactId>canal.common</artifactId>
    <version>1.1.4</version>
</dependency>
<dependency>
    <groupId>com.alibaba.otter</groupId>
    <artifactId>canal.protocol</artifactId>
    <version>1.1.4</version>
</dependency>
<dependency>
    <groupId>com.google.protobuf</groupId>
    <artifactId>protobuf-java</artifactId>
    <version>3.5.1</version>
</dependency>
Bean | 连接 canal.deployer
package tech.lixinlei.dianping.canal;

import com.alibaba.google.common.collect.Lists;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;

@Component
public class CanalClient implements DisposableBean{

    private CanalConnector canalConnector;

    @Bean
    public CanalConnector getCanalConnector(){
        canalConnector = CanalConnectors.newClusterConnector(Lists.newArrayList(
                new InetSocketAddress("127.0.0.1", 11111)),
                "example",
                "canal",
                "canal"
        );
        canalConnector.connect();
        // 指定filter，格式{database}.{table}，不传参数就是 subscribe 所有的内容
        canalConnector.subscribe();
        // 回滚寻找上次中断的位置
        canalConnector.rollback();
        
        return canalConnector;
    }

    /**
     * 容器销毁时调用
     * @throws Exception
     */
    @Override
    public void destroy() throws Exception {
        if(canalConnector != null){
            canalConnector.disconnect();
        }
    }

}
Bean | 定时从 canal.deployer 中读取 binlog 并解析、索引进 ElasticSearch
package tech.lixinlei.dianping.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import tech.lixinlei.dianping.dal.ShopModelMapper;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
public class CanalScheduling implements Runnable, ApplicationContextAware {

    private ApplicationContext applicationContext;

    @Autowired
    private ShopModelMapper shopModelMapper;

    @Resource
    private CanalConnector canalConnector;

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    @Override
    @Scheduled(fixedDelay = 100)
    public void run() {
        System.out.println("run");
        long batchId = -1;
        try{
            int batchSize = 1000;
            Message message = canalConnector.getWithoutAck(batchSize);
            batchId = message.getId();
            List<CanalEntry.Entry> entries = message.getEntries();
            if(batchId != -1 && entries.size() > 0){
                entries.forEach(entry -> {
                    if(entry.getEntryType() == CanalEntry.EntryType.ROWDATA){
                        // 解析处理
                        publishCanalEvent(entry);
                    }
                });
            }
            canalConnector.ack(batchId);
        }catch(Exception e){
            e.printStackTrace();
            canalConnector.rollback(batchId);
        }
    }

    /**
     * 将 binlog 中的一条（entry），
     * 解析成受影响的记录（change），再逐条解析受影响的记录（change），
     * 将记录（rowData）的数据结构从 List 转成 Map，
     * 完了交给 indexES 方式索引进 ElasticSearch；
     * @param entry binlog 中的一条；
     */
    private void publishCanalEvent(CanalEntry.Entry entry){
        CanalEntry.EventType eventType = entry.getHeader().getEventType();
        String database = entry.getHeader().getSchemaName();
        String table = entry.getHeader().getTableName();
        CanalEntry.RowChange change = null;
        try {
            change = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
            return;
        }
        change.getRowDatasList().forEach(rowData -> {
            List<CanalEntry.Column> columns = rowData.getAfterColumnsList();
            String primaryKey = "id";
            CanalEntry.Column idColumn = columns.stream().filter(column -> column.getIsKey()
                    && primaryKey.equals(column.getName())).findFirst().orElse(null);

            Map<String,Object> dataMap = parseColumnsToMap(columns);
            try{
                indexES(dataMap, database, table);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
    }

    Map<String,Object> parseColumnsToMap(List<CanalEntry.Column> columns){
        Map<String,Object> jsonMap = new HashMap<>();
        columns.forEach(column -> {
            if(column == null){
                return;
            }
            jsonMap.put(column.getName(), column.getValue());
        });
        return jsonMap;
    }

    private void indexES(Map<String,Object> dataMap, String database, String table) throws IOException {
        if(!StringUtils.equals("dianping", database)){
            return;
        }

        // result 查出来的记录是全字段，不像 canal.adapter 只能查出更改的字段；
        List<Map<String,Object>> result = new ArrayList<>();
        if(StringUtils.equals("seller", table)) {
            result = shopModelMapper.buildESQuery(new Integer((String)dataMap.get("id")), null, null);
        } else if (StringUtils.equals("category", table)){
            result = shopModelMapper.buildESQuery(null, new Integer((String)dataMap.get("id")), null);
        } else if (StringUtils.equals("shop", table)){
            result = shopModelMapper.buildESQuery(null, null, new Integer((String)dataMap.get("id")));
        } else {
            return;
        }

        // 调用 ES API 将 MySQL 中变化的数据索引进 ElasticSearch
        for(Map<String,Object> map : result){
            IndexRequest indexRequest = new IndexRequest("shop");
            indexRequest.id(String.valueOf(map.get("id")));
            indexRequest.source(map);
            restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
        }

    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

}
在 SpringBoot 层面打开定时任务的开关
@SpringBootApplication(scanBasePackages = {"tech.lixinlei.dianping"})
@MapperScan("tech.lixinlei.dianping.dal")
@EnableAspectJAutoProxy(proxyTargetClass = true)
@EnableScheduling
public class DianpingApplication {

    public static void main(String[] args) {
        SpringApplication.run(DianpingApplication.class, args);
    }

}
根据 binlog 中的内容，查出需要更新进 ElasticSearch 中的数据的 SQL
public interface ShopModelMapper {

    List<Map<String,Object>> buildESQuery(@Param("sellerId")Integer sellerId,
                                          @Param("categoryId")Integer categoryId,
                                          @Param("shopId")Integer shopId);

}

<select id="buildESQuery" resultType="java.util.Map">
  select a.id,a.name,a.tags,concat(a.latitude,',',a.longitude) as location,
  a.remark_score,a.price_per_man,a.category_id,b.name as category_name,a.seller_id,
  c.remark_score as seller_remark_score,c.disabled_flag as seller_disabled_flag
  from shop a inner join category b on a.category_id = b.id inner join seller c on c.id=a.seller_id
  <if test="sellerId != null">
    and c.id = #{sellerId}
  </if>
  <if test="categoryId != null">
    and b.id = #{categoryId}
  </if>
  <if test="shopId != null">
    and a.id = #{shopId}
  </if>
</select>
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 217,734评论 6赞 505
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 92,931评论 3赞 394
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 164,133评论 0赞 354
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 58,532评论 1赞 293
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 67,585评论 6赞 392
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 51,462评论 1赞 302
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 40,262评论 3赞 418
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 39,153评论 0赞 276
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 45,587评论 1赞 314
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 37,792评论 3赞 336
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 39,919评论 1赞 348
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 35,635评论 5赞 345
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 41,237评论 3赞 329
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,855评论 0赞 22
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,983评论 1赞 269
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 48,048评论 3赞 370
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 44,864评论 2赞 354