{"id":1679,"date":"2023-03-25T22:33:30","date_gmt":"2023-03-25T14:33:30","guid":{"rendered":"https:\/\/www.appblog.cn\/?p=1679"},"modified":"2023-04-23T21:48:51","modified_gmt":"2023-04-23T13:48:51","slug":"alibaba-canal-client-access","status":"publish","type":"post","link":"https:\/\/www.appblog.cn\/index.php\/2023\/03\/25\/alibaba-canal-client-access\/","title":{"rendered":"\u963f\u91cc\u5df4\u5df4Canal Client\u63a5\u5165"},"content":{"rendered":"

\u53c2\u8003\uff1ahttps:\/\/github.com\/alibaba\/canal\/wiki<\/a><\/p>\n

\n

\u9996\u5148\u542f\u52a8Canal Server<\/p>\n<\/blockquote>\n

\u521b\u5efamvn\u6807\u51c6\u5de5\u7a0b<\/h3>\n
mvn archetype:generate -DgroupId=com.alibaba.otter -DartifactId=canal.sample<\/code><\/pre>\n
<\/p>\n
pom.xml\u4f9d\u8d56\u914d\u7f6e<\/h3>\n
<dependency>\n    <groupId>com.alibaba.otter<\/groupId>\n    <artifactId>canal.client<\/artifactId>\n    <version>1.1.4<\/version>\n<\/dependency><\/code><\/pre>\n
ClientSample\u4ee3\u7801<\/h3>\n
package com.alibaba.otter.canal.sample;\n\nimport java.net.InetSocketAddress;\nimport java.util.List;\n\nimport com.alibaba.otter.canal.client.CanalConnectors;\nimport com.alibaba.otter.canal.client.CanalConnector;\nimport com.alibaba.otter.canal.common.utils.AddressUtils;\nimport com.alibaba.otter.canal.protocol.Message;\nimport com.alibaba.otter.canal.protocol.CanalEntry.Column;\nimport com.alibaba.otter.canal.protocol.CanalEntry.Entry;\nimport com.alibaba.otter.canal.protocol.CanalEntry.EntryType;\nimport com.alibaba.otter.canal.protocol.CanalEntry.EventType;\nimport com.alibaba.otter.canal.protocol.CanalEntry.RowChange;\nimport com.alibaba.otter.canal.protocol.CanalEntry.RowData;\n\npublic class SimpleCanalClientExample {\n\n    public static void main(String args[]) {\n        \/\/ \u521b\u5efa\u94fe\u63a5\n        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),\n                                                                                            11111), "test", "", "");\n        int batchSize = 1000;\n        int emptyCount = 0;\n        try {\n            connector.connect();\n            connector.subscribe(".*\\\\..*");\n            connector.rollback();\n            int totalEmptyCount = 120;\n            while (emptyCount < totalEmptyCount) {\n                Message message = connector.getWithoutAck(batchSize); \/\/ \u83b7\u53d6\u6307\u5b9a\u6570\u91cf\u7684\u6570\u636e\n                long batchId = message.getId();\n                int size = message.getEntries().size();\n                if (batchId == -1 || size == 0) {\n                    emptyCount++;\n                    System.out.println("empty count : " + emptyCount);\n                    try {\n                        Thread.sleep(1000);\n                    } catch (InterruptedException e) {\n                    }\n                } else {\n                    emptyCount = 0;\n                    \/\/ System.out.printf("message[batchId=%s,size=%s] \\n", batchId, size);\n                    printEntry(message.getEntries());\n                }\n\n                connector.ack(batchId); \/\/ \u63d0\u4ea4\u786e\u8ba4\n                \/\/ connector.rollback(batchId); \/\/ \u5904\u7406\u5931\u8d25, \u56de\u6eda\u6570\u636e\n            }\n\n            System.out.println("empty too many times, exit");\n        } finally {\n            connector.disconnect();\n        }\n    }\n\n    private static void printEntry(List<Entry> entrys) {\n        for (Entry entry : entrys) {\n            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {\n                continue;\n            }\n\n            RowChange rowChage = null;\n            try {\n                rowChage = RowChange.parseFrom(entry.getStoreValue());\n            } catch (Exception e) {\n                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),\n                                           e);\n            }\n\n            EventType eventType = rowChage.getEventType();\n            System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",\n                                             entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),\n                                             entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),\n                                             eventType));\n\n            for (RowData rowData : rowChage.getRowDatasList()) {\n                if (eventType == EventType.DELETE) {\n                    printColumn(rowData.getBeforeColumnsList());\n                } else if (eventType == EventType.INSERT) {\n                    printColumn(rowData.getAfterColumnsList());\n                } else {\n                    System.out.println("-------> before");\n                    printColumn(rowData.getBeforeColumnsList());\n                    System.out.println("-------> after");\n                    printColumn(rowData.getAfterColumnsList());\n                }\n            }\n        }\n    }\n\n    private static void printColumn(List<Column> columns) {\n        for (Column column : columns) {\n            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());\n        }\n    }\n}<\/code><\/pre>\n
\u8fd0\u884cClient<\/h3>\n
\u542f\u52a8Canal Client\u540e\uff0c\u53ef\u4ee5\u4ece\u63a7\u5236\u53f0\u770b\u5230\u7c7b\u4f3c\u6d88\u606f\uff1a<\/p>\n
empty count : 1\nempty count : 2\nempty count : 3\nempty count : 4<\/code><\/pre>\n
\u6b64\u65f6\u4ee3\u8868\u5f53\u524d\u6570\u636e\u5e93\u65e0\u53d8\u66f4\u6570\u636e<\/p>\n
\u89e6\u53d1\u6570\u636e\u5e93\u53d8\u66f4<\/h3>\n
mysql> use test;\nDatabase changed\nmysql> CREATE TABLE `xdual` (\n    ->   `ID` int(11) NOT NULL AUTO_INCREMENT,\n    ->   `X` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,\n    ->   PRIMARY KEY (`ID`)\n    -> ) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8 ;\nQuery OK, 0 rows affected (0.06 sec)\nmysql> insert into xdual(id,x) values(null,now());Query OK, 1 row affected (0.06 sec)<\/code><\/pre>\n
\u53ef\u4ee5\u4ece\u63a7\u5236\u53f0\u4e2d\u770b\u5230\uff1a<\/p>\n
empty count : 1\nempty count : 2\nempty count : 3\nempty count : 4\n================> binlog[mysql-bin.001946:313661577] , name[test,xdual] , eventType : INSERT\nID : 4    update=true\nX : 2013-02-05 23:29:46    update=true<\/code><\/pre>\n
\u66f4\u591a\u80fd\u529b<\/h3>\n
\u5982\u679c\u9700\u8981\u66f4\u8be6\u7ec6\u7684exmpale\u4f8b\u5b50\uff0c\u8bf7\u4e0b\u8f7dcanal\u5f53\u524d\u6700\u65b0\u6e90\u7801\u5305\uff0c\u91cc\u9762\u6709\u4e2aexample\u5de5\u7a0b<\/p>\n