阿里开源数据同步组件Canal_阿里canal

一.简介

canal是阿里开源的数据同步组件
这个是是git地址

二.使用步骤

1.安装配置mysql

1. 安装一个数据库（这个数据库是被监听的对象，我这里用的是mysql5.7）
2. 创建一个用户专门用于数据同步（这里也可以直接用root用户）

①创建用户 用户名：canal 密码：canal

create user 'canal'@'%' identified by 'canal';
1

②授予外部访问权限

grant SELECT, REPLICATION SLAVE, REPLICATION CLIENT on *.* to 'canal'@'%' identified by 'canal';
1

3. 修改mysql配置文件

# 打开binlog
log-bin=mysql-bin
# 选择ROW(行)模式
binlog-format=ROW
# 配置MySQL replaction需要定义，不要和canal的slaveId重复(canal1.1.4自动生成slaveId)
server_id=1
1
2
3
4
5
6

4. 修改完配置文件之后，重启mysql，再查看是否修改成功

show VARIABLES like 'log_bin';
1

5. 创建数据库

create database user character set utf8; 
1

6. 创建表

CREATE TABLE `user`.`tb_user`  (
  `id` int(32) NOT NULL,
  `username` varchar(32) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `age` int(32) NULL DEFAULT NULL,
  PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;
1
2
3
4
5
6

2.安装配置canal

1. 下载canal，我这里用的是1.1.4，github下载可能非常慢，这里再贴上我的百度云，提取码：ewji

2. 将canal放到/usr/local/canal目录下，解压

tar -zxvf canal.deployer-1.1.4.tar.gz 
1

3. 打开配置文件

vi conf/example/instance.properties
1

只需要修改这行代码配置文件就可以了，如果用户不是canal，还需要修改下dbUsername和dbPassword，
canal.instance.filter.regex不用修改，默认监听整个数据库，也可以写具体的表名，用逗号隔开

注意：如果这里使用的是docker安装的mysql，有可能没有vi命令，需要手动安装

apt-get update
1

apt-get install vim
1

如果上面的命令执行失败，我之前用centos 8的时候报错了，最后没有解决，更换成centos 7就可以了

3.启动canal

1. 启动canal

2. 查看日志文件，查看canal是否启动成功

启动成功

注意，这里启动的时候有可能会有很多问题出现

比如：数据库编码问题

如果是用navicat创建的数据库，这样看上去用的好像是utf8，但是在canal这里不行

我们可以看一下该数据库的编码格式，如果是这样的就会有问题

show variables like "%character%";
1

修改一下数据的编码格式

set character_set_client = utf8;

set character_set_server = utf8;

set character_set_connection = utf8;

set character_set_database = utf8;

set character_set_results = utf8;

set collation_connection = utf8_general_ci;

set collation_database = utf8_general_ci;

set collation_server = utf8_general_ci;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

这样再重启就不会有编码问题了

三.Java canal客户端

1. 引入maven依赖

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

1
2
3
4
5
6

2. 修改启动类

需要在项目启动时便持续监听

package com.xx;

import com.xx.client.CanalClient;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import javax.annotation.Resource;

/**
 * @author aqi
 * DateTime: 2021/2/24 4:43 下午
 * Description: canal启动类
 */
@SpringBootApplication
public class CanalApplication implements CommandLineRunner {

    @Resource
    private CanalClient canalClient;

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

    @Override
    public void run(String... args) throws Exception {
        // 项目启动,执行canal客户端监听
        canalClient.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
27
28
29
30
31

3. 编写canal客户端

这里操作数据库用的是JdbcTemplate

package com.xx.client;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @author aqi
 * DateTime: 2021/2/24 4:48 下午
 * Description: No Description
 */
@Component
public class CanalClient {

    //sql队列
    private Queue<String> SQL_QUEUE = new ConcurrentLinkedQueue<>();

    @Resource
    private JdbcTemplate jdbcTemplate;

    /**
     * canal入库方法
     */
    public void run() {
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("服务器ipd地址", 11111), "example", "", "");
        int batchSize = 1000;
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            try {
                while (true) {
                    //尝试从master那边拉去数据batchSize条记录，有多少取多少
                    Message message = connector.getWithoutAck(batchSize);
                    long batchId = message.getId();
                    int size = message.getEntries().size();
                    if (batchId == -1 || size == 0) {
                        Thread.sleep(1000);
                    } else {
                        dataHandle(message.getEntries());
                    }
                    connector.ack(batchId);

                    //当队列里面堆积的sql大于一定数值的时候就模拟执行
                    if (SQL_QUEUE.size() >= 1) {
                        executeQueueSql();
                    }
                }
            } catch (InterruptedException | InvalidProtocolBufferException e) {
                e.printStackTrace();
            }
        } finally {
            connector.disconnect();
        }
    }

    /**
     * 模拟执行队列里面的sql语句
     */
    public void executeQueueSql() {
        int size = SQL_QUEUE.size();
        for (int i = 0; i < size; i++) {
            String sql = SQL_QUEUE.poll();
            System.out.println("[sql]----> " + sql);

            this.execute(sql);
        }
    }

    /**
     * 数据处理
     */
    private void dataHandle(List<Entry> entrys) throws InvalidProtocolBufferException {
        for (Entry entry : entrys) {
            if (EntryType.ROWDATA == entry.getEntryType()) {
                RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
                EventType eventType = rowChange.getEventType();
                if (eventType == EventType.DELETE) {
                    saveDeleteSql(entry);
                } else if (eventType == EventType.UPDATE) {
                    saveUpdateSql(entry);
                } else if (eventType == EventType.INSERT) {
                    saveInsertSql(entry);
                }
            }
        }
    }

    /**
     * 保存更新语句
     */
    private void saveUpdateSql(Entry entry) {
        try {
            RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
            List<RowData> rowDatasList = rowChange.getRowDatasList();
            for (RowData rowData : rowDatasList) {
                List<Column> newColumnList = rowData.getAfterColumnsList();
                StringBuilder sql = new StringBuilder("update " + entry.getHeader().getTableName() + " set ");
                for (int i = 0; i < newColumnList.size(); i++) {
                    sql.append(" ").append(newColumnList.get(i).getName()).append(" = '").append(newColumnList.get(i).getValue()).append("'");
                    if (i != newColumnList.size() - 1) {
                        sql.append(",");
                    }
                }
                sql.append(" where ");
                List<Column> oldColumnList = rowData.getBeforeColumnsList();
                for (Column column : oldColumnList) {
                    if (column.getIsKey()) {
                        //暂时只支持单一主键
                        sql.append(column.getName()).append("=").append(column.getValue());
                        break;
                    }
                }
                SQL_QUEUE.add(sql.toString());
            }
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
    }

    /**
     * 保存删除语句
     */
    private void saveDeleteSql(Entry entry) {
        try {
            RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
            List<RowData> rowDatasList = rowChange.getRowDatasList();
            for (RowData rowData : rowDatasList) {
                List<Column> columnList = rowData.getBeforeColumnsList();
                StringBuilder sql = new StringBuilder("delete from " + entry.getHeader().getTableName() + " where ");
                for (Column column : columnList) {
                    if (column.getIsKey()) {
                        //暂时只支持单一主键
                        sql.append(column.getName()).append("=").append(column.getValue());
                        break;
                    }
                }
                SQL_QUEUE.add(sql.toString());
            }
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
    }

    /**
     * 保存插入语句
     */
    private void  saveInsertSql(Entry entry) {
        try {
            RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
            List<RowData> rowDatasList = rowChange.getRowDatasList();
            for (RowData rowData : rowDatasList) {
                List<Column> columnList = rowData.getAfterColumnsList();
                StringBuilder sql = new StringBuilder("insert into " + entry.getHeader().getTableName() + " (");
                for (int i = 0; i < columnList.size(); i++) {
                    sql.append(columnList.get(i).getName());
                    if (i != columnList.size() - 1) {
                        sql.append(",");
                    }
                }
                sql.append(") VALUES (");
                for (int i = 0; i < columnList.size(); i++) {
                    sql.append("'").append(columnList.get(i).getValue()).append("'");
                    if (i != columnList.size() - 1) {
                        sql.append(",");
                    }
                }
                sql.append(")");
                SQL_QUEUE.add(sql.toString());
            }
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
    }

    /**
     * 入库
     */
    public void execute(String sql) {
        jdbcTemplate.execute(sql);
    }
}

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
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193

注意：

如果项目启动失败，并且抛出canal连接失败，这个大概率是防火墙的问题，因为canal默认的端口号是11111，这个默认是关闭的，所以需要手动开启

查看已经开启的端口号

firewall-cmd --zone=public --list-ports
1

开启11111端口

firewall-cmd --zone=public --add-port=11111/tcp --permanent
1

重启防火墙

firewall-cmd --reload
1

4. 往数据库中插入数据

INSERT INTO tb_user (username, age) VALUES ('Tom', 20)
1

5. 查看数据是否同步，可以看到数据同步成功

三.总结

1. canal对于代码没有侵入性，基于监听mysql binlog日志去实现数据同步，实时性也非常的不错，这里有官方发出的性能测试结果，日常工作一般是够用了