`

【canal】canal hello world

阅读更多
具体参考:
https://github.com/alibaba/canal/wiki/QuickStart

准备
对于自建 MySQL , 需要先开启 Binlog 写入功能,配置 binlog-format 为 ROW 模式,my.cnf 中配置如下

[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义,不要和 canal 的 slaveId 重复
注意:针对阿里云 RDS for MySQL , 默认打开了 binlog , 并且账号默认具有 binlog dump 权限 , 不需要任何权限或者 binlog 设置,可以直接跳过这一步
授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant

CREATE USER canal IDENTIFIED BY 'canal'; 
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
FLUSH PRIVILEGES;
启动

官方指导的配置是有问题的,启动canal-server会报以下错误:
^@2020-06-29 08:27:52.320 [destination = example , address = /127.0.0.1:3306 , EventParser] ERROR c.a.otter.canal.parse.inbound.mysql.MysqlEventParser - dump address /127.0.0.1:3306 has an error, retrying. caused by
java.io.IOException: connect /127.0.0.1:3306 failure:java.io.IOException: Unexpected End Stream


log-bin=【有权限的绝对路径】 # 开启 binlog
重启mysql后生效




下载 canal, 访问 release 页面 , 选择需要的包下载, 如以 1.0.17 版本为例

wget https://github.com/alibaba/canal/releases/download/canal-1.0.17/canal.deployer-1.0.17.tar.gz
解压缩

mkdir /tmp/canal
tar zxvf canal.deployer-$version.tar.gz  -C /tmp/canal
解压完成后,进入 /tmp/canal 目录,可以看到如下结构
drwxr-xr-x 2 jianghang jianghang  136 2013-02-05 21:51 bin
drwxr-xr-x 4 jianghang jianghang  160 2013-02-05 21:51 conf
drwxr-xr-x 2 jianghang jianghang 1.3K 2013-02-05 21:51 lib
drwxr-xr-x 2 jianghang jianghang   48 2013-02-05 21:29 logs
配置修改

vi conf/example/instance.properties
## mysql serverId
canal.instance.mysql.slaveId = 1234
#position info,需要改成自己的数据库信息
canal.instance.master.address = 127.0.0.1:3306
canal.instance.master.journal.name =
canal.instance.master.position =
canal.instance.master.timestamp =
#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =
#username/password,需要改成自己的数据库信息
canal.instance.dbUsername = canal 
canal.instance.dbPassword = canal
canal.instance.defaultDatabaseName =
canal.instance.connectionCharset = UTF-8
#table regex
canal.instance.filter.regex = .\*\\\\..\*
canal.instance.connectionCharset 代表数据库的编码方式对应到 java 中的编码类型,比如 UTF-8,GBK , ISO-8859-1
如果系统是1个 cpu,需要将 canal.instance.parser.parallel 设置为 false
启动

sh bin/startup.sh
查看 server 日志

vi logs/canal/canal.log</pre>
2013-02-05 22:45:27.967 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## start the canal server.
2013-02-05 22:45:28.113 [main] INFO  com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[10.1.29.120:11111]
2013-02-05 22:45:28.210 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## the canal server is running now ......
查看 instance 的日志

vi logs/example/example.log
2013-02-05 22:50:45.636 [main] INFO  c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [canal.properties]
2013-02-05 22:50:45.641 [main] INFO  c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [example/instance.properties]
2013-02-05 22:50:45.803 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example
2013-02-05 22:50:45.810 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start successful....
关闭

sh bin/stop.sh


客户端:
package com.alibaba.test.steed.utils;

import java.net.InetSocketAddress;
import java.util.List;


import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;


public class SimpleCanalClientExample {


    public static void main(String args[]) {
        // 创建链接
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),
                11111), "example", "", "");
//        11111), "example", "", "");
        int batchSize = 1000;
        int emptyCount = 0;
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            int totalEmptyCount = 120;
            while (emptyCount < totalEmptyCount) {
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    emptyCount++;
                    System.out.println("empty count : " + emptyCount);
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                    }
                } else {
                    emptyCount = 0;
                    // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
                    printEntry(message.getEntries());
                }

                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }

            System.out.println("empty too many times, exit");
        } finally {
            connector.disconnect();
        }
    }

    private static void printEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }

            RowChange rowChage = null;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                        e);
            }

            EventType eventType = rowChage.getEventType();
            System.out.println(String.format("================&gt; binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType));

            for (RowData rowData : rowChage.getRowDatasList()) {
                if (eventType == EventType.DELETE) {
                    printColumn(rowData.getBeforeColumnsList());
                } else if (eventType == EventType.INSERT) {
                    printColumn(rowData.getAfterColumnsList());
                } else {
                    System.out.println("-------&gt; before");
                    printColumn(rowData.getBeforeColumnsList());
                    System.out.println("-------&gt; after");
                    printColumn(rowData.getAfterColumnsList());
                }
            }
        }
    }

    private static void printColumn(List<Column> columns) {
        for (Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
        }
    }

}


客户端消费:
empty count : 113
empty count : 114
================&gt; binlog[mysql-bin.000001:382] , name[example,tb_user] , eventType : INSERT
id : 1    update=false
aid : 1    update=false
name : 1111    update=false
passwd :     update=false
telephone :     update=false
icon :     update=false
email :     update=false
title :     update=false
status : 0    update=false
create_time : 2000-01-01 00:00:00    update=false
update_time : 2020-06-29 08:32:52    update=false
empty count : 1
empty count : 2
分享到:
评论

相关推荐

    canal客户端-canal.deployer-1.1.7-SNAPSHOT.tar.gz

    《深入理解Canal客户端:canal.deployer-1.1.7-SNAPSHOT.tar.gz剖析》 Canal,由阿里开源的一款高效、稳定的数据库增量日志订阅与消费组件,广泛应用于数据同步、实时数据接入等场景。在本篇文章中,我们将深入探讨...

    Linux下Canal安装和使用.doc

    Linux 下 Canal 安装和使用 Canal 是一个基于 MySQL 数据库增量日志解析的工具,提供增量数据订阅和消费。 Canal 的主要用途包括数据库镜像、数据库实时备份、索引构建和实时维护、业务 Cache 刷新、带业务逻辑的...

    canal.1.1.5包

    【标题】"canal.1.1.5包"指的是Canal项目的1.1.5版本的组件集合。Canal是阿里巴巴开源的一款基于MySQL数据库的数据变更发布与订阅系统,它能够实时捕获并同步数据库中的数据变化,常用于构建数据一致性、实时数据...

    canal.deployer-1.1.6

    《Canal Deployer 1.1.6:MySQL数据同步利器详解》 Canal Deployer是阿里巴巴开源的数据同步工具Canal的重要组成部分,主要用于部署和管理Canal实例。在1.1.6版本中,它提供了稳定高效的数据变更捕获与传输服务,...

    canal-canal-1.1.5-alpha-2.zip ;canal-server、adapter、admin

    canal-server(canal-deploy):可以直接监听MySQL的binlog,把自己伪装成MySQL的从库,只负责接收数据,并不做处理。 canal-adapter:相当于canal的客户端,会从canal-server中获取数据,然后对数据进行同步,可以...

    阿里canal组件:canal.deployer-1.1.7-SNAPSHOT.tar.gz

    阿里Canal是一款由阿里巴巴开源的、轻量级的数据变更订阅与推送组件,主要用来实现数据库的增量数据同步。在IT行业中,特别是在互联网公司,数据一致性是至关重要的,而Canal就是为了满足这种需求而诞生的。它能有效...

    canal 1.1.4 常用工具包

    canal.deployer-1.1.4.tar.gz canal.admin-1.1.4.tar.gz canal.adapter-1.1.4.tar.gz canal.example-1.1.4.tar.gz canal-canal-1.1.4.zip canal 全量工具包集合 https://juejin.cn/post/6864585112994971662

    canal-1.1.6

    【标题】"Canal-1.1.6" 是一个重要的数据库同步工具,主要用于实现MySQL数据的实时增量复制。在IT行业中,数据同步是确保多系统间数据一致性和实时性的关键环节,尤其在分布式系统、大数据处理和云计算环境中。Canal...

    springboot整合canal

    SpringBoot整合Canal是将阿里巴巴开源的数据库实时变更数据订阅与推送组件Canal与SpringBoot框架结合,以实现高效的数据同步。Canal主要用于数据库的实时增量数据同步,它能够监听MySQL数据库的binlog事件,当数据库...

    canal.adapter-1.1.6.zip

    【标题】"canal.adapter-1.1.6.zip" 提供的是Canal适配器的1.1.6版本,这是一个与数据库变更日志相关的软件组件。 【描述】"canal.adapter-1.1.6.zip" 指出这个压缩包包含的是Canal适配器的特定版本1.1.6,它是一个...

    canal-1.1.6-SNAPSHOT.rar

    【标题】"canal-1.1.6-SNAPSHOT.rar" 提供的是一个名为 Canal 的数据同步工具的特定版本,即 1.1.6-SNAPSHOT 版本。这个压缩包是针对 Java 开发环境,特别是 JDK 17 的安装文件。这意味着在使用前,你需要确保你的...

    canal.deployer-1.1.6.tar.gz

    《Canal Deployer 1.1.6:MySQL 数据库实时同步工具详解》 Canal Deployer 是阿里巴巴开源的一款高效、稳定、轻量级的 MySQL 数据库增量日志解析组件,其主要功能是实现 MySQL 数据库到其他系统(如 Elasticsearch...

    canal实现mysql到ES数据实时同步

    Canal包含三个主要组件:Canal Server、Canal Client和Canal Adapter。Canal Server负责监听MySQL的Binlog事件,并将这些事件转化为结构化的数据。Canal Client可以订阅并消费这些数据,而Canal Adapter则扮演着数据...

    canal-canal-1.0.22_源码

    【canal-canal-1.0.22_源码】是阿里巴巴开源的数据库实时增量数据订阅与同步工具,主要用于实现MySQL到其他数据库或者数据存储系统的实时数据迁移、同步,常用于构建数据仓库、日志收集等场景。Canal的核心功能是...

    canal.adapter-1.1.4.tar.gz

    《Canal Deployer 1.1.4:MySQL数据库监听利器》 Canal Deployer 是阿里巴巴开源的一个数据库增量日志订阅与消费组件,主要用于数据库的实时数据同步。在这个场景下,`canal.deployer-1.1.4.tar.gz` 文件扮演了关键...

    canal-1.1.5 全套

    包含canal全套资源包含以下 canal.adapter-1.1.5.tar.gz canal.admin-1.1.5.tar.gz canal.deployer-1.1.5.tar.gz canal.example-1.1.5.tar.gz

    canal.adapter-1.1.4.zip

    【标题】"Canal Adapter 1.1.4版本" Canal是阿里巴巴开源的一个数据库增量日志订阅与消费组件,主要用于实现数据库的实时数据同步。Canal Adapter是其配套的适配器模块,用于将Canal捕获到的数据库变更事件转化为...

    支持mysql5.6.19的canal版本

    MySQL的Canal是一款开源的数据库实时同步工具,它主要用于数据复制和数据变更事件的捕获。在本场景中,我们关注的是Canal与特定MySQL版本——5.6.19的兼容性问题。通常,数据库版本升级或降级可能会导致与现有同步...

    canal-1.1.5.zip

    《阿里巴巴开源项目Canal详解与应用》 Canal,由阿里巴巴开源的一款高性能、轻量级的MySQL数据库实时增量数据订阅和消费组件,主要用于实现数据库的数据变更实时同步到其他系统,如大数据处理、缓存更新等场景。在...

    canal安装包免费下载.mysql同步ES工具

    【标题】"Canal安装包免费下载:MySQL同步ES工具" 【描述】"这款工具无需复杂的配置,只需在服务器上解压即可使用,对于Windows用户同样便捷,无需积分即可获取。" 【知识点详解】 1. **Canal**: Canal是阿里...

Global site tag (gtag.js) - Google Analytics