华为云云耀云服务器L实例评测｜拉取创建canal镜像配置相关参数 & 搭建canal连接MySQL数据库 & spring项目应用canal初步

前言

最近华为云云耀云服务器L实例上新，也搞了一台来玩，本篇博客介绍如何在华为云上部署canal的docker镜像，以及在spring项目中的初步应用。

其他相关的华为云云耀云服务器L实例评测文章列表如下：

• 初始化配置SSH连接 & 安装MySQL的docker镜像 & 安装redis以及主从搭建 & 7.2版本redis.conf配置文件

• 安装Java8环境 & 配置环境变量 & spring项目部署 &【!】存在问题未解决

• 部署spring项目端口开放问题的解决 & 服务器项目环境搭建MySQL，Redis，Minio…指南

• 由于自己原因导致MySQL数据库被攻击 & MySQL的binlog日志文件的理解

引出

---

1.Canal:基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费；
2.Canal的使用，MySQL配置，docker的Canal安装；
3.云服务器开放端口，测试服务器端口是否开放的方式；
4.spring应用canal的初步；

一、认识Canal管道

1.canal是什么？

https://github.com/alibaba/canal

https://github.com/alibaba/canal/wiki/ClientExample

Canal是阿里开源的一款基于Mysql binlog的增量订阅和消费组件，通过它可以订阅数据库的binlog日志，然后进行一些数据消费，如数据镜像、数据异构、数据索引、缓存更新等。相对于消息队列，通过这种机制可以实现数据的有序化和一致性。

canal [kə’næl]，译意为水道/管道/沟渠，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费

早期阿里巴巴因为杭州和美国双机房部署，存在跨机房同步的业务需求，实现方式主要是基于业务 trigger 获取增量变更。从 2010 年开始，业务逐步尝试数据库日志解析获取增量变更进行同步，由此衍生出了大量的数据库增量订阅和消费业务。

基于日志增量订阅和消费的业务包括

• 数据库镜像
• 数据库实时备份
• 索引构建和实时维护(拆分异构索引、倒排索引等)
• 业务 cache 刷新
• 带业务逻辑的增量数据处理

当前的 canal 支持源端 MySQL 版本包括 5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x

2.canal的原理初步

MySQL主从的原理

• MySQL master 将数据变更写入二进制日志( binary log, 其中记录叫做二进制日志事件binary log events，可以通过 show binlog events 进行查看)
• MySQL slave 将 master 的 binary log events 拷贝到它的中继日志(relay log)
• MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据

canal的原理----> 伪装成从

• canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议
• MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )
• canal 解析 binary log 对象(原始为 byte 流)

canal的应用展望

• canal从MySQL拿到数据，然后同步到redis上，就不需要进行延迟双删来保证MySQL和Redis的数据一致性，避免延迟双删出现的其他问题。

二、MySQL容器创建canal用户

docker exec -it mysql_3306 bash
mysql -uroot -p    
show VARIABLES like 'log_%';
1
2
3

create user 'canal'@'%' IDENTIFIED with mysql_native_password by 'canal';
grant SELECT, REPLICATION SLAVE, REPLICATION CLIENT on *.* to 'canal'@'%';
FLUSH PRIVILEGES;
1
2
3

三、拉取canal镜像创建容器

1.查询拉取canal镜像

docker pull canal/canal-server
1

root@hcss-ecs-52b8:~# docker pull canal/canal-server
Using default tag: latest
latest: Pulling from canal/canal-server
1c8f9aa56c90: Pull complete 
c5e21c824d1c: Pull complete 
4ba7edb60123: Pull complete 
80d8e8fac1be: Pull complete 
705a43657e98: Pull complete 
28e38bfb6fe7: Pull complete 
7d51a00deff6: Pull complete 
4f4fb700ef54: Pull complete 
Digest: sha256:0d1018759efd92ad331c7cc379afa766c8d943ef48ef8d208ade646f54bf1565
Status: Downloaded newer image for canal/canal-server:latest
docker.io/canal/canal-server:latest
1
2
3
4
5
6
7
8
9
10
11
12
13
14

2.运行canal容器，获取配置文件

为了后续挂载启动做准备工作

docker run --name canal -itd canal/canal-server
1

docker run

• -i：以交互模式运行容器
• -t：为容器重新分配一个伪输入终端
• –name :容器名称
• –privileged: 设置容器公开权限（默认为true）
• -p :映射端口 linux端口: 容器内置端口(mysql默认端口为3306)
• -v : linux挂载文件夹/文件和容器内路径的映射
• -e: 容器的环境变量（设置mysql默认用户名&密码)
• -d: 后台运行容器，并返回容器ID

docker exec -it canal bash
1

/home/admin/canal-server/conf/canal.properties
/home/admin/canal-server/conf/example/instance.properties
1
2

docker cp canal:/home/admin/canal-server/conf/canal.properties ./
docker cp canal:/home/admin/canal-server/conf/example/instance.properties ./
1
2

从canal的docker容器中拷贝配置文件

拷贝配置文件的结果

3.编辑canal配置文件

vim instance.properties 
1

编辑配置文件所需的参数，mysql的内部ip地址，binlog文件名以及日志的位置

docker inspect mysql_3306 | grep IPA
show master status;
1
2

注意不要忘记加端口号

修改连接MySQL数据库的用户名和密码

4.删掉之前的canal容器，创建挂载启动容器

docker stop canal 
docker rm canal
1
2

docker run -itd  --name canal \
-p 11111:11111 --privileged=true \
-v /usr/local/software/canal/conf/instance.properties:/home/admin/canal-server/conf/example/instance.properties \
-v /usr/local/software/canal/conf/canal.properties:/home/admin/canal-server/conf/example/canal.properties \
canal/canal-server
1
2
3
4
5

5.查看日志

docker logs canal
1

查看日志，运行成功

6.开放canal的端口

firewall-cmd --zone=public --add-port=11111/tcp --permanent
firewall-cmd --reload
firewall-cmd --zone=public --list-ports
1
2
3

ps：需要在华为云后台也开放一下端口

tips：监听服务端口是否开发的方法

nc 即 netcat。netcat 是一个简单的 Unix 工具，它使用 TCP 或 UDP 协议去读写网络连接间的数据。

它被设计成为一个可信赖的后端工具，可被直接使用或者简单地被其他程序或脚本调用。

与此同时，它也是一个富含功能的网络调试和探索工具，因为它可以创建你所需的几乎所有类型的连接，并且还拥有几个内置的有趣功能。

netcat 有三类功能模式，它们分别为连接模式、监听模式和隧道模式。

nc（netcat）命令的一般语法：

$ nc [-options] [HostName or IP] [PortNumber]
1

命令详解：

• nc：即执行的命令主体；
• z：零 I/O 模式（被用来扫描）；
• v：显式地输出；
• w3：设置超时时间为 3 秒；
• 192.168.1.8：目标系统的 IP 地址；
• 22：需要验证的端口。

使用案例

[root@localhost conf]# nc -zvw3 124.80.139.65 3927
Ncat: Version 7.50 ( https://nmap.org/ncat )
Ncat: Connection timed out.
[root@localhost conf]# nc -zvw3 124.80.139.65 3927
Ncat: Version 7.50 ( https://nmap.org/ncat )
Ncat: Connected to 124.80.139.65:3927.
Ncat: 0 bytes sent, 0 bytes received in 0.02 seconds.
1
2
3
4
5
6
7

四、springboot整合canal

1.引入依赖

        
        <dependency>
            <groupId>com.alibaba.ottergroupId>
            <artifactId>canal.clientartifactId>
            <version>1.1.0version>
        dependency>
1
2
3
4
5
6

2.改造官网的案例代码

package com.woniu.fresh.config.redis;


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;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

/**
 * 用canal管道监听MySQL数据变化，自动更新redis缓存
 */
@Slf4j
@Component
public class AutoUpdateRedis {

    @Value("${canal.host}")
    private String host;

    @Value("${canal.port}")
    private Integer port;

    public void run() {

        // 创建链接
        final InetSocketAddress HOST = new InetSocketAddress(host,port);
//        final InetSocketAddress HOST = new InetSocketAddress("192.168.111.130",11111);
        CanalConnector connector = CanalConnectors.newSingleConnector(HOST, "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 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("================> 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 {
                    // 修改
                    log.debug("-------修改之前before");
                    updateBefore(rowData.getBeforeColumnsList());
                    log.debug("-------修改之后after");
                    updateAfter(rowData.getAfterColumnsList());
                }
            }
        }
    }

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

    /**
     * 数据库更新之前
     * @param columns
     */
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    private  void updateBefore(List<Column> columns) {
        for (Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
//            // 如果数据更新，就更新缓存的数据
//            if ("username".equals(column.getName())){
//                // 把更新之前的数据删除
//                stringRedisTemplate.opsForSet().remove("usernames", column.getValue());
//                break;
//            }
        }
    }
    private  void updateAfter(List<Column> columns) {
        for (Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
//            // 如果数据更新，就更新缓存的数据
//            if ("username".equals(column.getName()) && column.getUpdated()){
//                // 把更新后的数据放入缓存
//                stringRedisTemplate.opsForSet().add("usernames", column.getValue());
//                break;
//            }
        }
    }
}


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

3.主启动类启动canal

package com.woniu.fresh;

import com.woniu.fresh.config.redis.AutoUpdateRedis;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.CommandLineRunner;

@SpringBootApplication
@Slf4j
public class FreshApp implements CommandLineRunner {
    public static void main(String[] args) {
        SpringApplication.run(FreshApp.class);
    }

    @Autowired
    private AutoUpdateRedis autoUpdateRedis;

    @Override
    public void run(String... args) throws Exception {
        log.debug(">>>>>启动缓存自动更新");
        autoUpdateRedis.run();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

4.启动canal并修改数据库

5.后台监听到变化

---

总结

1.Canal:基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费；
2.Canal的使用，MySQL配置，docker的Canal安装；
3.云服务器开放端口，测试服务器端口是否开放的方式；
4.spring应用canal的初步；