1. 程式人生 > 實用技巧 >docker部署canal並實現簡單的資料同步

docker部署canal並實現簡單的資料同步

1.首先docker拉去canal映象,這裡不加版本號,預設為latest

docker pull canal/canal-server

2.檢視映象是否拉去成功

docker images

3.拉去完成後,先啟動下canal,主要是為了從裡面copy出配置檔案

#啟動映象 
docker run --name canal -d canal/canal-server
#進入容器 檢視配置檔案路徑

 docker exec -it canal bash

#找到檔案位置後 exit退出容器 將容器內部檔案copy到外部
 docker cp canal:
/home/admin/canal-server/conf/canal.properties /home/canal
 docker cp canal:
/home/admin/canal-server/conf/example/instance.properties /home/canal

4.檔案copy完成後主要是修改instance這個檔案。第一個紅框是你需要監聽資料庫的地址和埠;第二個紅框是你資料庫的使用者和密碼,這個使用者資訊一定是要有全部許可權的使用者,非root使用者;第三個是匹配資料表的規則,我這裡預設為全部表

5.修改完成後,將之前的canal容器關閉,重新起一個新的容器.

#關閉容器
docker stop canal

#移除容器
docker rm canal

#啟動新的 這裡-v是將外部的檔案掛載到容器內部 這樣就不用每次啟動都要配置引數了
docker run --name canal -p 11111:11111 -d -v /home/canal/instance.properties:/home/admin/canal-server/conf/example/instance.properties -v /home/canal/canal.properties:/home/admin/canal-server/conf/canal.properties canal

6. 這樣canal基本的配置完成了,但是mysql也需要配置,這裡簡單說明下。對mysql中 my.cnf檔案進行修改,開啟binglog。需要注意的是server_id和canal中canal.properties中的id一定要不同,切記

7.使用java程式碼來進行測試,這裡是依賴

8. 這裡需要注意的是,黃色背景程式碼。我這裡源資料庫和目標資料庫都在一個伺服器,canal配置的監聽的是所有表。所以如果不加判斷的話,增刪改都會反覆執行。如果你也是監聽所有表,也需要對庫名進行判斷。

package com.atguigu.canal.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.apache.commons.dbutils.DbUtils; import org.apache.commons.dbutils.QueryRunner; import org.springframework.stereotype.Component; import javax.annotation.Resource; import javax.sql.DataSource; import java.net.InetSocketAddress; import java.sql.Connection; import java.sql.SQLException; import java.util.List; import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; @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("你的伺服器ip地址", 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); } } /** * 資料處理 * * @param entrys */ private void dataHandle(List<Entry> entrys) throws InvalidProtocolBufferException { for (Entry entry : entrys) { if(entry.getHeader().getSchemaName().equals("hc")){ return; } 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); } } }

9. 現在可以啟動啟動類,然後對資料庫進行操作,就會出現效果了。

寫的不是特別仔細 希望大家多多擔待