canal使用入坑,親測 !!!!
原來用windows本地運行,在本地模擬客戶端和服務端,鼓搗了一天都沒運行成功。。。最後放棄了,改用
windows的客戶端+ubuntu的服務端(客戶端程序運行在window,canal監聽運行在Ubuntu中)
用到:windows+ubuntu+eclipse+mysql+canal
1、客戶端(windows)
1)用eclipse建一個java項目,建類
public class ClientTest { public static void main(String args[]) { // 創建鏈接 這個ip是你虛擬機的ip
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("10.168.12.43",11111), "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; printEntry(message.getEntries()); } connector.ack(batchId); // 提交確認 } System.out.println("empty too many times, exit"); } finally { connector.disconnect(); } } private static 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 { System.out.println("-------> before"); printColumn(rowData.getBeforeColumnsList()); System.out.println("-------> after"); printColumn(rowData.getAfterColumnsList()); } } } } private static void printColumn(List<Column> columns) { for (Column column : columns) { System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated()); } } }
2)引入依賴(這裏我用的1.0.24最新的是1.1.3)
<dependency> <groupId>com.alibaba.otter</groupId> <artifactId>canal.client</artifactId> <version>1.0.24</version> </dependency>
簡單吧!!!
2、服務端
主要工作在這裏了!!!
1、確定你的虛擬機安裝了java環境和mysql數據庫
2、查看是否啟用了日誌:
mysql>show variables like ‘log_bin‘;
3、開啟binlog
如果log_bin關閉,需要在etc下面找到my.cnf,開啟binlog:
server_id=1 log-bin=/var/lib/mysql/mysql-bin
4、添加canal mysql數據庫賬號
CREATE USER canal IDENTIFIED BY ‘canal‘; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO ‘canal‘@‘%‘; GRANT ALL PRIVILEGES ON *.* TO ‘canal‘@‘%‘ ; FLUSH PRIVILEGES;
5、下載1.0.24的(https://github.com/alibaba/canal/releases/tag/canal-1.0.24)這個包 canal.deployer-1.0.24.tar.gz
6、修改 vi canal/conf/example/instance.properties
## mysql serverId
canal.instance.mysql.slaveId = 1234
# position info
canal.instance.master.address = 10.168.12.43:3306
canal.instance.master.journal.name =mysql-bin.000003
canal.instance.master.position =
canal.instance.master.timestamp =
……
canal.instance.dbUsername = canal
canal.instance.dbPassword = canal
canal.instance.defaultDatabaseName =testcanal
canal.instance.connectionCharset = UTF-8
# table regex
canal.instance.filter.regex = .*\\..*
ip不要寫錯了,這個是虛擬機的ip,因為要監聽的是運行與Ubuntu上的數據庫(這裏用的mysql)的操作。
testcanal是默認數據庫,這個是你要監聽的那個表所屬的數據庫
7、啟動canal
sh bin/startup.sh
8、查看日誌
vi logs/canal/canal.log vi logs/example/example.log
看看日誌是否啟動成功沒報錯
9、測試
1)確保startup.sh 文件運行了
2)在windows上運行上面的ClientTest 程序
3)在mysql的testcanal數據庫中建立表,進行增刪改查
4)會在eclipse上打印操作數據庫的信息
參考:https://www.cnblogs.com/janes/p/9318576.html
canal使用入坑,親測 !!!!