基於阿里的Canal實現資料同步
阿新 • • 發佈：2020-12-30
一、開啟同步資料庫的binlog功能
（1）開啟同步資料端的資料庫服務（比如我的將一號虛擬機器上的mysql資料庫作為同步操作資料庫）
systemctl start mysql.service
mysql -h192.168.137.100 -uroot -p123
（2）檢查mysql的binlog功能是否開啟（可見是OFF，關閉的）
MySQL [(none)]> show variables like 'log_bin';
+---------------+-------+
| Variable_name | Value |
+---------- （3）開啟binlog功能，修改my.cnf配置檔案：vim /etc/my.cnf
在檔案中新增如下內容後，重啟mysql服務，在檢視mysql的binlog功能。可見已經開啟。
log-bin=mysql-bin
binlog_format=ROW
server_id=1
MySQL [(none)]> show variables like 'log_bin';
+---------------+-------+
| Variable_name | Value | 二、安裝canal工具
（1）下載地址：https://github.com/alibaba/canal/releases
（2）下載檔案：canal.deployer-1.1.4.tar.gz
（3）檔案傳到 /usr/local/canal資料夾下，解壓
tar zxvf canal.deployer-1.1.4.tar.gz
（4）修改配置檔案vi conf/example/instance.properties
修改如下內容為資料庫連線：
canal.instance.master.address=192.168.137.100:3306
canal.instance.dbPassword=123
（5）啟動canal資料同步工具 cd bin
[[email protected] bin]# ./startup.sh
三、測試專案準備
（1）新建專案，引入如下依賴
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!--mysql-->
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
</dependency>
<dependency>
<groupId>commons-dbutils</groupId>
<artifactId>commons-dbutils</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
</dependency>
（2）專案配置檔案配置本地同步資料庫
# 服務埠
server.port=1000
# 服務名
spring.application.name=canal-client
# 環境設定：dev、test、prod
spring.profiles.active=dev
# mysql資料庫連線
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.url=jdbc:mysql://192.168.137.130:3306/TeachOnLine?useUnicode=true&useSSL=false&characterEncoding=utf8&serverTimezone=Asia/Shanghai
spring.datasource.username=root
spring.datasource.password=123
（3）建立如下啟動類
@SpringBootApplication
public class CanalApplication implements CommandLineRunner {
@Resource
private CanalClient canalClient;
// canalClient為資料庫同步操作執行類
public static void main(String[] args) {
SpringApplication.run(CanalApplication.class, args);
}
@Override
public void run(String... strings) throws Exception {
//專案啟動，執行canal客戶端監聽
canalClient.run();
}
}
（4）建立CanalClient資料庫同步操作執行類內容如下（記得保證192.168.137.100伺服器的11111埠可通）
@Component
public class CanalClient {
//sql佇列
private Queue<String> SQL_QUEUE = new ConcurrentLinkedQueue<>();
@Resource
private DataSource dataSource;
/**
* canal入庫方法
*/
public void run() {
// 建立遠端資料庫連線物件
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.137.100",
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 e) {
e.printStackTrace();
} catch (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.toString());
}
}
/**
* 資料處理
*
* @param entrys
*/
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);
}
}
}
}
/**
* 儲存更新語句
*
* @param 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();
StringBuffer sql = new StringBuffer("update " + entry.getHeader().getTableName() + " set ");
for (int i = 0; i < newColumnList.size(); i++) {
sql.append(" " + newColumnList.get(i).getName()
+ " = '" + newColumnList.get(i).getValue() + "'");
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() + "=" + column.getValue());
break;
}
}
SQL_QUEUE.add(sql.toString());
}
} catch (InvalidProtocolBufferException e) {
e.printStackTrace();
}
}
/**
* 儲存刪除語句
*
* @param entry
*/
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();
StringBuffer sql = new StringBuffer("delete from " + entry.getHeader().getTableName() + " where ");
for (Column column : columnList) {
if (column.getIsKey()) {
//暫時只支援單一主鍵
sql.append(column.getName() + "=" + column.getValue());
break;
}
}
SQL_QUEUE.add(sql.toString());
}
} catch (InvalidProtocolBufferException e) {
e.printStackTrace();
}
}
/**
* 儲存插入語句
*
* @param entry
*/
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();
StringBuffer sql = new StringBuffer("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("'" + columnList.get(i).getValue() + "'");
if (i != columnList.size() - 1) {
sql.append(",");
}
}
sql.append(")");
SQL_QUEUE.add(sql.toString());
}
} catch (InvalidProtocolBufferException e) {
e.printStackTrace();
}
}
/**
* 入庫
* @param sql
*/
public void execute(String sql) {
Connection con = null;
try {
if(null == sql) return;
con = dataSource.getConnection();
QueryRunner qr = new QueryRunner();
int row = qr.execute(con, sql);// 語句最終執行！
System.out.println("update: "+ row);
} catch (SQLException e) {
e.printStackTrace();
} finally {
DbUtils.closeQuietly(con);
}
}
}
四、測試
（1）啟動類開啟，在192.168.137.100(遠端)資料庫TeachOnLine的表members中插入資料
insert into members values(2,‘candy’,22);
（2）檢視控制檯列印資訊
[sql]----> insert into members (id,username,age) VALUES ('2','candy','22')
update: 1
（3）在192.168.137.130(本地)資料庫TeachOnLine的表members中出現相同資料。
注：在此例中本地資料庫會同步遠端資料庫的變化，反過來修改本地資料庫內容，遠端資料庫不會同步本地資料庫的加粗樣式變化