Rumah >Java >javaTutorial >Bagaimana SpringBoot menyepadukan kaedah Canal
(1.1.5 telah banyak berubah, pelanggan di sini menggunakan 1.1.4)
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.2.2.RELEASE</version> <relativePath/> <!-- lookup parent from repository --> </parent> <groupId>top.yueshushu</groupId> <artifactId>learn</artifactId> <version>1.0-SNAPSHOT</version> <name>Canal</name> <description>学习 Canal</description> <properties> <java.version>1.8</java.version> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> </dependency> <!-- 导入配置文件处理器,配置文件进行绑定就会有提示,需要重启 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-configuration-processor</artifactId> <optional>true</optional> </dependency> <!--导入自动热步署的依赖--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-devtools</artifactId> <optional>true</optional> </dependency> <!--引入MySql的驱动--> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> </dependency> <!--引入springboot与mybatis整合的依赖--> <dependency> <groupId>org.mybatis.spring.boot</groupId> <artifactId>mybatis-spring-boot-starter</artifactId> <version>2.1.4</version> </dependency> <!-- 引入pagehelper分页插件 --> <dependency> <groupId>com.github.pagehelper</groupId> <artifactId>pagehelper-spring-boot-starter</artifactId> <version>1.2.5</version> </dependency> <!--添加 druid-spring-boot-starter的依赖的依赖--> <dependency> <groupId>com.alibaba</groupId> <artifactId>druid-spring-boot-starter</artifactId> <version>1.1.14</version> </dependency> <!--SpringBoot 的aop 模块--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-aop</artifactId> </dependency> <!--添加canal的依赖. 重要. 使用 1.1.4--> <dependency> <groupId>com.alibaba.otter</groupId> <artifactId>canal.client</artifactId> <version>1.1.4</version> </dependency> <dependency> <groupId>joda-time</groupId> <artifactId>joda-time</artifactId> <version>2.9.4</version> </dependency> </dependencies> <build> <!--将该目录下的文件全部打包成类的路径--> <resources> <resource> <directory>src/main/resources</directory> </resource> </resources> <plugins> <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> </plugin> </plugins> </build> </project>
/** * 一个简单的canal 的连接测试程序 */ @Test public void connectionTest() { //1. 创建连接 填充对应的地址信息 ,要监控的实例和相应的用户名和密码 CanalConnector canalConnector = CanalConnectors.newSingleConnector( new InetSocketAddress( "127.0.0.1", 11111 ), "example", "canal", "canal" ); //2. 进行连接 canalConnector.connect(); log.info(">>>连接成功:{}", canalConnector); }
17:26:32.179 [utama] INFO atas.yueshushu.learn.CanalDemoTest - >>>Sambungan berjaya: com.alibaba.otter.canal.client.impl . SimpleCanalConnector@31ef45e3
/** * 获取数据信息. 可以发现,未获取到数据 . 这个应该是实时的. */ @Test public void getDataTest() { //1. 创建连接 CanalConnector canalConnector = CanalConnectors.newSingleConnector( new InetSocketAddress("127.0.0.1", 11111), "example", "canal", "canal" ); // 进行连接 canalConnector.connect(); //3. 注册,看使用哪个数据库表 canalConnector.subscribe("springboot.user"); //4. 获取 1条数据 Message message = canalConnector.get(1); log.info("获取的数据:id:{},数据:{}", message.getId(), message); if (message.getId() == -1) { log.info(">>>未获取到数据"); return; } //5. 获取相应的数据集合 List<CanalEntry.Entry> entries = message.getEntries(); for (CanalEntry.Entry entry : entries) { log.info(">>>获取数据 {}", entry); //获取表名 CanalEntry.Header header = entry.getHeader(); log.info(">>>获取表名:{}", header.getTableName()); CanalEntry.EntryType entryType = entry.getEntryType(); log.info(">>获取类型 {}:,对应的信息:{}", entryType.getNumber(), entryType.name()); //获取数据 ByteString storeValue = entry.getStoreValue(); log.info(">>>输出存储的值:{}", storeValue); } }
Masukkan sekeping data dalam perpustakaan utama
rreeeLaksanakan semula :
insert into springboot.user(id,name,age,sex,description) values(1,'canal添加用户',24,'男','学习canal');
Anda boleh mendapatkan maklumat perubahan data yang sepadan pada bila-bila masa.
Anda akan mendapati bahawa nilai storeValue sukar untuk ditafsirkan. Data ini perlu dihuraikan.
/** * 获取数据信息. 获取现在的数据. 再次执行时,就没有这个数据了. */ @Test public void getNowDataTest() { //1. 创建连接 CanalConnector canalConnector = CanalConnectors.newSingleConnector( new InetSocketAddress("127.0.0.1", 11111), "example", "canal", "canal" ); // 进行连接 canalConnector.connect(); //3. 注册,看使用哪个数据库表 canalConnector.subscribe("springboot.user"); for (;;) { //4. 获取 1条数据 Message message = canalConnector.get(1); log.info("获取的数据:id:{},数据:{}", message.getId(), message); if (message.getId() == -1) { log.info(">>>未获取到数据"); try { TimeUnit.MILLISECONDS.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } continue; } //5. 获取相应的数据集合 List<CanalEntry.Entry> entries = message.getEntries(); for (CanalEntry.Entry entry : entries) { log.info(">>>获取数据 {}", entry); //获取表名 CanalEntry.Header header = entry.getHeader(); log.info(">>>获取表名:{}", header.getTableName()); CanalEntry.EntryType entryType = entry.getEntryType(); log.info(">>获取类型 {}:,对应的信息:{}", entryType.getNumber(), entryType.name()); //获取数据 ByteString storeValue = entry.getStoreValue(); log.info(">>>输出存储的值:{}", storeValue); } } }
Laksanakan sql sekali lagi
/** * 将 storeValue 进行解析,解析成我们能看懂的语句. * 对数据库 cud 进行处理操作观看一下. * 发现,点是不好的,也有多余的记录信息. * * @throws Exception 异常 */ @Test public void convertDataTest() throws Exception { //1. 创建连接 CanalConnector canalConnector = CanalConnectors.newSingleConnector( new InetSocketAddress("127.0.0.1", 11111), "example", "canal", "canal" ); //2. 进行连接 canalConnector.connect(); canalConnector.subscribe("springboot.user"); for (;;) { //获取信息 Message message = canalConnector.get(1); if (message.getId() == -1L) { // log.info("未获取到数据"); try { TimeUnit.MILLISECONDS.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } continue; } List<CanalEntry.Entry> entryList = message.getEntries(); //对获取到的数据进行处理 log.info(">>获取到{}条数据", entryList.size()); for (CanalEntry.Entry entry : entryList) { CanalEntry.Header header = entry.getHeader(); log.info(">>>获取表名:{}", header.getTableName()); //获取类型. CanalEntry.EntryType entryType = entry.getEntryType(); log.info(">>类型编号 {},类型名称:{}", entryType.getNumber(), entryType.name()); //获取存入日志的值 ByteString storeValue = entry.getStoreValue(); //将这个值进行解析 CanalEntry.RowChange rowChange = RowChange.parseFrom(storeValue); String sql = rowChange.getSql(); log.info(">>>获取对应的sql:{}", sql); // 这个sql 可能是 批量的sql语句 List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList(); for (CanalEntry.RowData rowData : rowDatasList) { log.info(">>>获取信息:{}", rowData); //对数据进行处理 List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList(); List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList(); beforeColumnsList.forEach( n -> log.info("哪个列{},原先是{},是否被更新{}", n.getName(), n.getValue(), n.getUpdated()) ); afterColumnsList.forEach( n -> log.info("哪个列{},后来是{},是否被更新{}", n.getName(), n.getValue(), n.getUpdated()) ); } } } }
Temui Jenis lain seperti: TRANSACTIONBEGIN juga diproses
insert into springboot.user(id,name,age,sex,description) values(2,'canal添加用户2',25,'男','学习canal2');
Sisipan, kemas kini dan pemadaman masing-masing diproses
Mulakan program ujian dahulu:
Tidak mencetak sebarang maklumat. Jadual utama melaksanakan pernyataan tambah:/** * 类型转换数据 * * @throws Exception 异常 */ @Test public void dataTypeTest() throws Exception { CanalConnector canalConnector = CanalConnectors.newSingleConnector( new InetSocketAddress( "127.0.0.1", 11111 ), "example", "canal", "canal" ); canalConnector.connect(); canalConnector.subscribe("springboot.user"); for(;;){ Message message = canalConnector.get(1); if (message.getId() == -1) { TimeUnit.SECONDS.sleep(1); continue; } List<CanalEntry.Entry> entries = message.getEntries(); for (CanalEntry.Entry entry : entries) { CanalEntry.EntryType entryType = entry.getEntryType(); //只要 RowData 数据类型的 if (!CanalEntry.EntryType.ROWDATA.equals(entryType)) { continue; } String tableName = entry.getHeader().getTableName(); log.info(">>>对表 {} 进行操作", tableName); ByteString storeValue = entry.getStoreValue(); RowChange rowChange = RowChange.parseFrom(storeValue); //行改变 CanalEntry.EventType eventType = rowChange.getEventType(); switch (eventType) { case INSERT: { insertHandler(rowChange); break; } case UPDATE: { updateHandler(rowChange); break; } case DELETE: { deleteHandler(rowChange); break; } default: { break; } } } } } private void deleteHandler(RowChange rowChange) { log.info(">>>>执行删除的方法"); List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList(); for (CanalEntry.RowData rowData : rowDatasList) { List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList(); for (CanalEntry.Column column : beforeColumnsList) { log.info(">>>>>字段 {} 删除数据 {}", column.getName(), column.getValue()); } } } private void updateHandler(RowChange rowChange) { log.info(">>>执行更新的方法"); List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList(); for (CanalEntry.RowData rowData : rowDatasList) { List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList(); List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList(); Map<String, String> beforeValueMap = beforeColumnsList.stream().collect( Collectors.toMap( CanalEntry.Column::getName, CanalEntry.Column::getValue ) ); Map<String, String> afterValueMap = afterColumnsList.stream().collect( Collectors.toMap( CanalEntry.Column::getName, CanalEntry.Column::getValue ) ); beforeValueMap.forEach((column, beforeValue) -> { String afterValue = afterValueMap.get(column); Boolean update = beforeValue.equals(afterValue); log.info("修改列:{},修改前的值:{},修改后的值:{},是否更新:{}", column, beforeValue, afterValue, update); }); } } /** * 插入数据. 只有后的数据. * * @param rowChange 行改变 */ private void insertHandler(RowChange rowChange) { log.info(">>>执行添加 的方法"); List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList(); for (CanalEntry.RowData rowData : rowDatasList) { List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList(); for (CanalEntry.Column column : afterColumnsList) { if (!StringUtils.hasText(column.getValue())) { continue; } log.info("字段 {} 插入了数据 {}", column.getName(), column.getValue()); } } }dan maklumat akan dicetak: Ini sangat boleh dibaca.
Jadual utama melakukan operasi pengubahsuaian.
insert into springboot.user(id,name,age,sex,description) values(4,'canal添加用户4',25,'男','学习canal4');Apabila mengemas kini, jika setiap medan adalah sama seperti sebelumnya, tiada penggunaan log akan dijana. Jadual utama melaksanakan operasi pemadaman:
update springboot.user set name='开开心心',age=26,description='岳泽霖' where id =4;Pemerolehan di atas semuanya diperolehi satu data pada satu masa . Kecekapan agak rendahMendapatkan berbilang keping data pada satu masa
delete from springboot.user where id =4;
/** * 一次性获取多条数据。 * sql 执行多条。 */ @Test public void dataMoreTest() throws Exception { //1. 创建 canal连接对象 CanalConnector canalConnector = CanalConnectors.newSingleConnector( new InetSocketAddress( "127.0.0.1", 11111 ), "example", "canal", "canal" ); canalConnector.connect(); // 订阅哪个对象 canalConnector.subscribe("springboot.user"); for (; ; ) { // Message message = canalConnector.get(3, 5L, TimeUnit.SECONDS); Message message = canalConnector.get(3); if (message.getId() == -1) { // 未获取到数据 continue; } List<CanalEntry.Entry> entries = message.getEntries(); for (CanalEntry.Entry entry : entries) { CanalEntry.EntryType entryType = entry.getEntryType(); if (!CanalEntry.EntryType.ROWDATA.equals(entryType)) { continue; } String tableName = entry.getHeader().getTableName(); log.info(">>>>对表{} 执行操作", tableName); CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); //对类型进行处理 CanalEntry.EventType eventType = rowChange.getEventType(); switch (eventType) { case INSERT: { insertHandler(rowChange); break; } case UPDATE: { updateHandler(rowChange); break; } case DELETE: { deleteHandler(rowChange); break; } default: { break; } } } } } private void deleteHandler(CanalEntry.RowChange rowChange) { log.info(">>>>执行删除的方法"); List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList(); for (CanalEntry.RowData rowData : rowDatasList) { List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList(); for (CanalEntry.Column column : beforeColumnsList) { log.info(">>>>>字段 {} 删除数据 {}", column.getName(), column.getValue()); } } } private void updateHandler(CanalEntry.RowChange rowChange) { log.info(">>>执行更新的方法"); List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList(); for (CanalEntry.RowData rowData : rowDatasList) { List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList(); List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList(); Map<String, String> beforeValueMap = beforeColumnsList.stream().collect( Collectors.toMap( CanalEntry.Column::getName, CanalEntry.Column::getValue ) ); Map<String, String> afterValueMap = afterColumnsList.stream().collect( Collectors.toMap( CanalEntry.Column::getName, CanalEntry.Column::getValue ) ); beforeValueMap.forEach((column, beforeValue) -> { String afterValue = afterValueMap.get(column); Boolean update = beforeValue.equals(afterValue); log.info("修改列:{},修改前的值:{},修改后的值:{},是否更新:{}", column, beforeValue, afterValue, update); }); } } /** * 插入数据. 只有后的数据. * * @param rowChange 行改变 */ private void insertHandler(CanalEntry.RowChange rowChange) { log.info(">>>执行添加 的方法"); List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList(); for (CanalEntry.RowData rowData : rowDatasList) { List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList(); for (CanalEntry.Column column : afterColumnsList) { if (!StringUtils.hasText(column.getValue())) { continue; } log.info("字段 {} 插入了数据 {}", column.getName(), column.getValue()); } } }.get(3) bermakna mendapat 3 rekod pada satu masa.canalConnector.get(3, 5L, TimeUnit.SECONDS); mencapai 5 saat. Maklumat paparan kesan adalah sama seperti sebelumnya, jadi ia tidak akan ditunjukkan lagi. maklumat konfigurasi ack
// Message message = canalConnector.get(3, 5L, TimeUnit.SECONDS); Message message = canalConnector.get(3);Maklumat utama:
Mesej mesej = canalConnector.getWithoutAck(3, 2L, TimeUnit.SECONDS);
//Rollback//canalConnector.rollback();//Sahkan konfigurasi ack canalConnector.ack(message.getId());Sahkan secara manual bahawa mesej telah digunakan.Apabila mesej rollback() ditarik balik, mesej akan digunakan semula.canalConnector.rollback();
Pernyataan pelaksanaan:
/** * 一次性获取多条数据。 * sql 执行多条。 */ @Test public void dataMoreTest() throws Exception { //1. 创建 canal连接对象 CanalConnector canalConnector = CanalConnectors.newSingleConnector( new InetSocketAddress( "127.0.0.1", 11111 ), "example", "canal", "canal" ); canalConnector.connect(); // 订阅哪个对象 canalConnector.subscribe("springboot.user"); for (; ; ) { Message message = canalConnector.getWithoutAck(3, 2L, TimeUnit.SECONDS); if (message.getId() == -1) { // 未获取到数据 TimeUnit.MILLISECONDS.sleep(500); continue; } log.info(">>>>获取对应的 id: {}",message.getId()); List<CanalEntry.Entry> entries = message.getEntries(); for (CanalEntry.Entry entry : entries) { CanalEntry.EntryType entryType = entry.getEntryType(); if (!CanalEntry.EntryType.ROWDATA.equals(entryType)) { continue; } String tableName = entry.getHeader().getTableName(); log.info(">>>>对表{} 执行操作", tableName); CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); //对类型进行处理 CanalEntry.EventType eventType = rowChange.getEventType(); switch (eventType) { case INSERT: { insertHandler(rowChange); break; } case UPDATE: { updateHandler(rowChange); break; } case DELETE: { deleteHandler(rowChange); break; } default: { break; } } } //进行回滚 // canalConnector.rollback(); //确认ack 配置 canalConnector.ack(message.getId()); } } private void deleteHandler(CanalEntry.RowChange rowChange) { log.info(">>>>执行删除的方法"); List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList(); for (CanalEntry.RowData rowData : rowDatasList) { List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList(); for (CanalEntry.Column column : beforeColumnsList) { log.info(">>>>>字段 {} 删除数据 {}", column.getName(), column.getValue()); } } } private void updateHandler(CanalEntry.RowChange rowChange) { log.info(">>>执行更新的方法"); List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList(); for (CanalEntry.RowData rowData : rowDatasList) { List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList(); List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList(); Map<String, String> beforeValueMap = beforeColumnsList.stream().collect( Collectors.toMap( CanalEntry.Column::getName, CanalEntry.Column::getValue ) ); Map<String, String> afterValueMap = afterColumnsList.stream().collect( Collectors.toMap( CanalEntry.Column::getName, CanalEntry.Column::getValue ) ); beforeValueMap.forEach((column, beforeValue) -> { String afterValue = afterValueMap.get(column); Boolean update = beforeValue.equals(afterValue); log.info("修改列:{},修改前的值:{},修改后的值:{},是否更新:{}", column, beforeValue, afterValue, update); }); } } /** * 插入数据. 只有后的数据. * * @param rowChange 行改变 */ private void insertHandler(CanalEntry.RowChange rowChange) { log.info(">>>执行添加 的方法"); List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList(); for (CanalEntry.RowData rowData : rowDatasList) { List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList(); for (CanalEntry.Column column : afterColumnsList) { if (!StringUtils.hasText(column.getValue())) { continue; } log.info("字段 {} 插入了数据 {}", column.getName(), column.getValue()); } } }
Jika ia ditukar kepada pengesahan manual,
canalConnector.ack(message.getId()); hanya akan digunakan sekali.Atas ialah kandungan terperinci Bagaimana SpringBoot menyepadukan kaedah Canal. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!