Docker：Canal安装和配置（第二种方法）（推荐）

待我称王封你为后i 2024-03-22 09:40 13阅读 0赞

### Canal安装和配置 ###

*  **创建mysql容器**

docker run -id --name canal_mysql \
    -v /mnt/canal_mysql:/var/lib/mysql \
    -p 3306:3306 \
    -e MYSQL_ROOT_PASSWORD=123456 mysql:5.7
    
    docker ps

*  **安装vim**

需要在MySQL容器中修改配置文件，但是容器中默认没有vim命令，需要进行安装。

直接执行命令安装vim速度会很慢，因为使用的是国外的源，需要更新Debian源以提高速度。

#在宿主机创建sources.list配置文件 在mnt目录下
    vi sources.list
    #内容为：
    deb http://mirrors.tuna.tsinghua.edu.cn/debian/ buster main
    deb http://mirrors.tuna.tsinghua.edu.cn/debian-security buster/updates main
    deb http://mirrors.tuna.tsinghua.edu.cn/debian/ buster-updates main
    
    #复制宿主机的配置到MySQL容器中
    docker cp sources.list canal_mysql:/etc/apt/
    
    #进入MySQL容器
    docker exec -it canal_mysql /bin/bash
    
    #执行安装命令
    apt-get update && apt-get install vim -y

*  **修改MySQL配置**

需要让canal伪装成salve并正确获取mysql中的binary log，首先要开启 Binlog 写入功能，配置 binlog-format 为 ROW 模式，命令如下：

#修改MySQL配置文件
    vim /etc/mysql/mysql.conf.d/mysqld.cnf
    #添加的内容如下：
    log-bin=mysql-bin
    binlog-format=ROW
    server_id=1
    #开启binlog 选择ROW模式 
    #server_id不要和canal的slaveId重复

*  **重启MySQL**

docker restart canal_mysql

远程登录MySQL，查看配置状态，执行以下sql：

show variables like 'log_bin';
    show variables like 'binlog_format';
    show master status;

*  **创建Canal账号**

创建连接MySQL的账号`canal`并授予作为 MySQL slave 的权限，执行以下sql：

# 创建账号
    CREATE USER canal IDENTIFIED BY 'canal'; 
    # 授予权限
    GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
    # 刷新并应用
    FLUSH PRIVILEGES;

*  **创建canal-server容器**

docker run -d --name canal-server \
    -p 11111:11111 canal/canal-server:v1.1.4

*  **配置canal-server**

#进入canal-server容器
    docker exec -it canal-server /bin/bash
    #编辑canal-server的配置 
    vi canal-server/conf/example/instance.properties

内容如下：  
![在这里插入图片描述][8e23815554724a60aa250af57fd57a4b.png_pic_center]

*  **重启canal-server**

修改完成后重启canal-server，并查看日志：

#按ctrl+D退出容器，并重启容器
    docker restart canal-server
    #重启成功后进入容器
    docker exec -it canal-server /bin/bash
    #查看日志
    tail -100f canal-server/logs/example/example.log

*  **简单使用**

在数据库服务中创建数据库canal\_test并创建表：

CREATE TABLE `student` (
      `id` varchar(20) NOT NULL,
      `name` varchar(50) DEFAULT NULL,
      `age` int(11) DEFAULT NULL,
      `sex` varchar(5) DEFAULT NULL,
      `city` varchar(20) DEFAULT NULL,
      PRIMARY KEY (`id`)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

创建Maven工程canal-demo，在pom.xml中添加依赖：

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

编写代码获取canal数据：

import com.alibaba.otter.canal.client.CanalConnector;
    import com.alibaba.otter.canal.client.CanalConnectors;
    import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
    import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
    import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
    import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
    import com.alibaba.otter.canal.protocol.Message;
    import java.net.InetSocketAddress;
    import java.util.concurrent.TimeUnit;
    
    public class CanalTest {
        
    
        public static void main(String[] args) {
        
            String ip = "39.98.107.235"; // canal server 的ip
            String destination = "example";
            //创建连接对象
            CanalConnector canalConnector = CanalConnectors.newSingleConnector(
                    new InetSocketAddress(ip, 11111), destination, "", ""
            );
    
            //进行连接
            canalConnector.connect();
            //进行订阅
            canalConnector.subscribe();
    
            int batchSize = 5 * 1024;
            //使用死循环不断的获取canal信息
            while (true) {
        
                //获取Message对象
                Message message = canalConnector.getWithoutAck(batchSize);
                long id = message.getId();
                int size = message.getEntries().size();
    
                System.out.println("当前监控到的binLog消息数量是：" + size);
    
                //判断是否有数据
                if (id == -1 || size == 0) {
        
                    //如果没有数据，等待1秒
                    try {
        
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                } else {
        
                    //如果有数据，进行数据解析
                    List<Entry> entries = message.getEntries();
    
                    //遍历获取到的Entry集合
                    for (Entry entry : entries) {
        
                        System.out.println("----------------------------------------");
                        System.out.println("当前的二进制日志的条目（entry）类型是：" + entry.getEntryType());
    
                        //如果属于原始数据ROWDATA，进行打印内容
                        if (entry.getEntryType() == EntryType.ROWDATA) {
        
                            try {
        
                                //获取存储的内容
                                RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
    
                                //打印事件的类型，增删改查哪种 eventType
                                System.out.println("事件类型是：" + rowChange.getEventType());
    
                                //打印改变的内容(增量数据)
                                for (RowData rowData : rowChange.getRowDatasList()) {
        
                                    System.out.println("改变前的数据：" + rowData.getBeforeColumnsList());
                                    System.out.println("改变后的数据：" + rowData.getAfterColumnsList());
                                }
    
                            } catch (Exception e) {
        
                                e.printStackTrace();
                            }
                        }
                    }
                    //消息确认已经处理了
                    canalConnector.ack(id);
                }
            }
        }
    }

数据对象格式：

Entry
        Header
            version         [协议的版本号,default = 1]
            logfileName     [binlog文件名]
            logfileOffset   [binlog position]
            serverId        [服务端serverId]
            serverenCode    [变更数据的编码]
            executeTime     [变更数据的执行时间]
            sourceType      [变更数据的来源,default = MYSQL]
            schemaName      [变更数据的schemaname]
            tableName       [变更数据的tablename]
            eventLength     [每个event的长度]
            eventType       [insert/update/delete类型,default = UPDATE]
            props           [预留扩展]
            gtid            [当前事务的gitd]
        entryType           [事务头BEGIN/事务尾END/数据ROWDATA/HEARTBEAT/GTIDLOG]
        storeValue          [byte数据,可展开，对应的类型为RowChange]    
    RowChange
        tableId             [tableId,由数据库产生]
        eventType           [数据变更类型,default = UPDATE]
        isDdl               [标识是否是ddl语句，比如create table/drop table]
        sql                 [ddl/query的sql语句]
        rowDatas            [insert/update/delete的变更数据，可为多条，event可对应多条变更，如批处理]
            beforeColumns   [字段信息，增量数据(修改前,删除前),Column类型的数组]
            afterColumns    [字段信息，增量数据(修改后,新增后),Column类型的数组] 
            props           [预留扩展]
        props               [预留扩展]
        ddlSchemaName       [ddl/query的schemaName，会存在跨库ddl，需要保留执行ddl的当前schemaName]
    Column 
        index               [字段下标]      
        sqlType             [jdbc type]
        name                [字段名称(忽略大小写)，在mysql中是没有的]
        isKey               [是否为主键]
        updated             [是否发生过变更]
        isNull              [值是否为null]
        props               [预留扩展]
        value               [字段值,timestamp,Datetime是一个时间格式的文本]
        length              [对应数据对象原始长度]
        mysqlType           [字段mysql类型]

[8e23815554724a60aa250af57fd57a4b.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/7d7a96661c1545d6b874db3248986fc4.png