delete from springboot.user where id =4;
Bagaimana SpringBoot menyepadukan kaedah Canal
May 13, 2023 am 09:22 AMpom.xml Tambah pergantungan canal.client
(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>
Pemprosesan fungsi perniagaan
Program Sambungan Mudah
/** * 一个简单的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
Dapatkan data sekali
/** * 获取数据信息. 可以发现,未获取到数据 . 这个应该是实时的. */ @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 :
Gelung untuk mendapatkan data
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.
Menghuraikan nilai storeValue
/** * 获取数据信息. 获取现在的数据. 再次执行时,就没有这个数据了. */ @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()) ); } } } }
Jenis yang berbeza diproses secara berbeza
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:
/** * 类型转换数据 * * @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()); } } }
Jadual utama melakukan operasi pengubahsuaian.
insert into springboot.user(id,name,age,sex,description) values(4,'canal添加用户4',25,'男','学习canal4');
update springboot.user set name='开开心心',age=26,description='岳泽霖' where id =4;
delete from springboot.user where id =4;
Salin selepas log masuk
Titik pengubahsuaian: 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()); } } }
// Message message = canalConnector.get(3, 5L, TimeUnit.SECONDS); Message message = canalConnector.get(3);
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()); } } }Salin selepas log masukJika 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!

Artikel Panas

Alat panas Tag

Artikel Panas

Tag artikel panas

Notepad++7.3.1
Editor kod yang mudah digunakan dan percuma

SublimeText3 versi Cina
Versi Cina, sangat mudah digunakan

Hantar Studio 13.0.1
Persekitaran pembangunan bersepadu PHP yang berkuasa

Dreamweaver CS6
Alat pembangunan web visual

SublimeText3 versi Mac
Perisian penyuntingan kod peringkat Tuhan (SublimeText3)

Topik panas

Bagaimana Springboot menyepadukan Jasypt untuk melaksanakan penyulitan fail konfigurasi

Cara menggunakan Redis untuk melaksanakan kunci teragih dalam SpringBoot

Bagaimana SpringBoot menyepadukan Redisson untuk melaksanakan baris gilir kelewatan

Analisis perbandingan dan perbezaan antara SpringBoot dan SpringMVC

Bagaimana untuk menyelesaikan masalah bahawa springboot tidak boleh mengakses fail selepas membacanya ke dalam pakej balang

Bagaimana SpringBoot menyesuaikan Redis untuk melaksanakan penyirian cache

Bagaimana untuk mendapatkan nilai dalam application.yml dalam springboot

Bagaimana untuk melaksanakan Springboot+Mybatis-plus tanpa menggunakan pernyataan SQL untuk menambah berbilang jadual
