spring+mysql集成canal

淩亂°似流年 2023-06-29 15:59 39阅读 0赞

# spring+mysql集成canal #

每天多学一点点~  
话不多说，这就开始吧…

### 文章目录 ###

*  spring+mysql集成canal
 *  1.前言
 *  2. 全量索引与增量索引介绍
 *  3. mysql集成canal
 *  4.spring集成canal
 *  5.效果演示
 *  6.结语

# 1.前言 #

博主现在的项目用到了es，但是之前的工程都已经写好了，不想改原本代码，但是数据库数据修改的时候，es的索引也要跟着改变，那怎么办呢？？？请看下文介绍

往期专题博客

*  [Centos7安装elasticsearch及kibana并进行安全设置][Centos7_elasticsearch_kibana]
 *  [ElasticSearch集群搭建][ElasticSearch]
 *  [elk和efk的搭建][elk_efk]

# 2. 全量索引与增量索引介绍 #

1.  全量索引构建  
    从头全部重新建。  
    一般分情况：如果是文本类的检索 因为其数据量庞大不会轻易的全量更新索引，一般以月为单位重建索引。  
    如果是电商类更新较为频繁又要求实时检索的系统 一般以天为单位进行重建索引。博主这边是交通部的项目，以月就行。  
    什么样的情况下要建全量？迁移 copy服务器。  
    Shard更改，第一次建数据，增减es字段，分词修改了，时间久了增量可能会丢数据，提高我们的查全率。  
    **数据量庞大的系统**：采取大数据处理平台进行离线索引的重建一般会部署两套检索系统，一套用来备份。当要进行全量更新时就更新这个备份索引，更新完成后将和正在使用的索引进行切换。  
    **数据量较小的系统**：使用常规多线程处理就可以了。但也要建立一个备份的索引，只是不用在部署一套es集群，一般就在一个集群里面建两个索引，当全量更新完后也对之进行切换，本次项目中就是使用这种。  
    以上方案的好处就是不管我重建索引有多慢或者失败了 都不会影响用户的使用，对之来说是无感的。
2.  增量索引构建  
    只建或者修改更新的数据  
    (1) 增量构建索引的标准  
    1.准实时性：数据变更后 es也需要马上变更，否则就影响了用户体验。  
    2.性能要求：快，有些像电商高并发  
    3.高可用&实现简单  
    (2) 增量构建索引的可选方案：  
    1.**单系统中**：一般比较简单，插入数据时直接更新es数据  
    2.**分布式系统**：搜索中心作为单独部署的，大部分情况应该就是这种  
    2.1 利用成熟的消息中间件：Mq，Kafka等。通常的做法就是业务系统有数据变更的时候发出消息，搜索中心监听到后进行数据的变更。这种方案的缺点就是比较依赖其他业务系统。  
    2.2 通过数据库层面的更新：此种方式也有两种，通过sql查询语句定时扫描数据库（updatetime，只适合第一次的全量），此种方式不灵活很难把握这个更新的间隔度。还有一种就是采用阿里巴巴推出的**canal**中间件，这种秩序配置数据库即可，对原业务代码无需操作，接下来介绍这种方式。

# 3. mysql集成canal #

1.下载  
[githup下载地址][githup] 选择合成自己的版本，本文采用的是1.1.3版本。（阿里果然是爸爸，每个java人心中的圣地233333333）  
canal.deployer-1.1.3.tar 是用来做增量的，全量可以用canal.adapter。不过博主这边链表操作比较多，所以一般全量都是用代码的。至于canal的介绍，去githup上自己看吧，这里贴一张官方图

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70]

2.开启mysql的binlog  
注意Mysql版本不要低于**5.6**,低于5.6的会出现你意想不到的异常。本文选用 10.2.26-MariaDB(反正高于mysql5.6就行)  
（1）查看binlog是否开启：show variables like '%log\_bin%’;  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 1]  
很显然，需要开启log\_bin ~  
（2）找到mysql 的配置文件添加以下内容：通常是my.ini(windows)或者 my.cnf(linux)文件

find / -name my.cnf  #找到my.cnf文件
    vim my.cnf #修改
    service mysqld restart   #重启mysql
    #加入 以下 配置
    server_id = 1
    binlog_format = ROW		# 行模式
    log_bin = mysql_bin.log		
    # 这两个可以不配置
    expire-logs-days = 14
    max-binlog-size = 500M

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 2]（3）确认binlog开启成功  
show variables like ‘%log\_bin%’; 下图所示就表示开启成功

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 3]

3.配置canal账户  
因为canal就是默认的账户，这里先配置一下，后面直接用  
配置一个账户，能够有**replication**权限

（1）创建canal账户 密码也是canal

CREATE USER canal IDENTIFIED BY 'canal';

（2）赋权限

GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';

（3）GRANT ALL PRIVILEGES ON *.* TO ‘canal’@’%’; \#也可以直接赋予所有权限，看自己需求  
（4）FLUSH PRIVILEGES; \#刷新权限配置

4.解压

tar -xvf canal.deployer-1.1.3.tar

我这里的路径是 /usr/local/elasticsearch/canal  
去config目录下配置信息  
（1）canal.properties文件：配置端口信息，其余的可以不用改 这里我全部用了默认配置

（2） 将conf目录下的example列子复制一份，并取名为自己需要的，博主这是book。

cp -r example/ /usr/local/elasticsearch/canal/conf/book

（3）修改book文件夹下的instance.properties  
配置数据库的地址和用户名密码

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 4]  
salveId 随便写一个  
![在这里插入图片描述][20200113145851464.png]  
canal正是我们前面所创建的用户

![在这里插入图片描述][2020011314590683.png]  
这里可改改不改。默认是把所有的表都监听

![在这里插入图片描述][20200113145934331.png]  
这里没用到mq，就先注释了

（4）启动

./bin/start.sh   #这样并不代表真的启动成功

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 5]

（5）去 logs/book 目录下查看日志

cd /usr/local/elasticsearch/canal/logs/book
    tail -1000f  book.log

查看启动日志，确认成功

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 6]  
到这里为止，cannal配置完成~

# 4.spring集成canal #

1.引入jar包

<dependency>
       <groupId>com.alibaba.otter</groupId>
       <artifactId>canal.client</artifactId>
       <version>1.1.3</version>
    </dependency>

2.注入spring

@Component
    public class CanalClient implements DisposableBean{ 
    
        private CanalConnector canalConnector;
    
        @Bean
        public CanalConnector getCanalConnector(){ 
            //即使我们用的是集群，其实同时也只有一个canal能工作，只有当他挂了 才会启用其他的，所以就是一个备份
        	canalConnector = CanalConnectors.newClusterConnector(Lists.newArrayList(
               new InetSocketAddress("192.168.73.132", 11111)),
                    "book","canal","canal"
               );
            canalConnector.connect();
            //指定filter，格式{database}.{table}
            canalConnector.subscribe("*.*");//指定我们要监听的表 这是监听所有
            //如果要监听的表 很多，这里最好建议用多线程，每一个canal服务只处理几张表就行,
            //canalConnector.subscribe("test.read_book_pd"); // 表示监听 test库下read_book_pd"表
            //回滚寻找上次中断的为止
            canalConnector.rollback();
            return canalConnector;
        }
        @Override
        public void destroy() throws Exception { 
            if(canalConnector != null){ 
                canalConnector.disconnect();
            }
        }
    }

3.canal集成es

@Component
    @Slf4j
    public class CanalScheduling implements Runnable{ 
    
       
    
        @Resource
        private CanalConnector canalConnector;
        @Resource
        private ReadBookPdMapper readBookPdMapper;
    
        @Autowired
        private RestHighLevelClient restHighLevelClient;
    
        @Override
        @Scheduled(fixedDelay = 100)  // 每100ms读取数据
        public void run() { 
            long batchId = -1;
            try { 
                int batchSize = 1000;    //一次取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);        //失败的话进行回滚
            }
    
        }
    
        private void publishCanalEvent(CanalEntry.Entry entry) { 
            log.info("收到canal消息{}", entry.toString());
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) { 
                return;
            }
            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;
            }
            EventType eventType = change.getEventType();
            change.getRowDatasList().forEach(rowData -> { 
    
                List<CanalEntry.Column> columns = null;
                if (eventType == EventType.DELETE)                    //对于es来说 只要关注 delete 和 update 还有insert
                    columns = rowData.getBeforeColumnsList();        //为什么这里是before 因为 相对于delete之后，不就是空了么，所以拿之前的一个
                else
                    columns = rowData.getAfterColumnsList();        //其他的都素hiafter
                Map<String, Object> dataMap = parseColumnsToMap(columns);        //解析成map 格式
                try { 
                    indexES(dataMap, database, table, eventType);    //真正的去改es
                } 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, EventType eventType)
                throws IOException { 
            try { 
                if (eventType == EventType.DELETE) { 
                    log.info("删除索引Id={},type={},value={}", dataMap.get("id"), eventType.toString(), dataMap.toString());
                    DeleteRequest deleteRequest = new DeleteRequest("book", "_doc", String.valueOf(dataMap.get("id")));
                    restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
                } else { 
    
    
                    //这里有两种方式
    
                    /** * 1. 直接拿canal过来的数据-->针对单表 */
    /* log.info("方式1，直接拿canal数据--> 更新索引Id={},type={},value={}", dataMap.get("id"), eventType.toString(), dataMap.toString()); IndexRequest indexRequest = new IndexRequest("book", "_doc"); indexRequest.id(String.valueOf(dataMap.get("id"))); indexRequest.source(dataMap); restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);*/
    
                    /** * 2. 拿主键id去db查询--->针对关联表 */
    				List<Map<String, Object>> result = readBookPdMapper
    						.buildESQuery(new Integer((String) dataMap.get("id")));
    				for (Map<String, Object> map : result) { 
    					//如果是又业务关联的 这里就要写自己的业务代码
    					log.info("方式2，去db查(或者调接口)--> 更新索引Id={},type={},value={}", map.get("id"), eventType.toString(), map.toString());
    
    					IndexRequest indexRequest = new IndexRequest("book", "_doc");
    					indexRequest.id(String.valueOf(map.get("id")));
    					indexRequest.source(map);
    					restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
    				}
    
    
                }
    
            } catch (Exception e) { 
                e.printStackTrace();
            }
    
        }
    }

# 5.效果演示 #

之前已经把mysql中数据导入到es中，现在对mysql中的数据进行crud操作，cannal都能感知到。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 7]  
再去es看一下，发现数据已经更新  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 8]

# 6.结语 #

**世上无难事，只怕有心人，每天积累一点点，fighting！！！**

[Centos7_elasticsearch_kibana]: https://blog.csdn.net/weixin_42437633/article/details/103749055
[ElasticSearch]: https://blog.csdn.net/weixin_42437633/article/details/103789688
[elk_efk]: https://blog.csdn.net/weixin_42437633/article/details/104053530
[githup]: https://github.com/alibaba/canal/releases
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200113162155426.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200113144727649.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/2020011314485082.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200113144955538.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200113145821899.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw==,size_16,color_FFFFFF,t_70
[20200113145851464.png]: https://img-blog.csdnimg.cn/20200113145851464.png
[2020011314590683.png]: https://img-blog.csdnimg.cn/2020011314590683.png
[20200113145934331.png]: https://img-blog.csdnimg.cn/20200113145934331.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20200113150032689.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20200113150111282.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20200113161924610.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/20200113161948957.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw==,size_16,color_FFFFFF,t_70