Jadual Kandungan
pom.xml Tambah pergantungan canal.client
Pemprosesan fungsi perniagaan
Program Sambungan Mudah
Dapatkan data sekali
Gelung untuk mendapatkan data
Menghuraikan nilai storeValue
Jenis yang berbeza diproses secara berbeza
delete from springboot.user where id =4;
Salin selepas log masuk
" >
delete from springboot.user where id =4;
Salin selepas log masuk
Rumah Java javaTutorial Bagaimana SpringBoot menyepadukan kaedah Canal

Bagaimana SpringBoot menyepadukan kaedah Canal

May 13, 2023 am 09:22 AM
springboot canal

pom.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>
Salin selepas log masuk

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);
    }
Salin selepas log masuk

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);
        }
    }
Salin selepas log masuk

Bagaimana SpringBoot menyepadukan kaedah Canal

Masukkan sekeping data dalam perpustakaan utama

rreee

Laksanakan semula :

Bagaimana SpringBoot menyepadukan kaedah Canal

Gelung untuk mendapatkan data

insert into springboot.user(id,name,age,sex,description) values(1,&#39;canal添加用户&#39;,24,&#39;男&#39;,&#39;学习canal&#39;);
Salin selepas log masuk

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);
            }
        }
    }
Salin selepas log masuk

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())
                    );
                }
            }
        }
    }
Salin selepas log masuk

Bagaimana SpringBoot menyepadukan kaedah Canal

Jenis yang berbeza diproses secara berbeza

Temui Jenis lain seperti: TRANSACTIONBEGIN juga diproses

insert into springboot.user(id,name,age,sex,description) 
values(2,&#39;canal添加用户2&#39;,25,&#39;男&#39;,&#39;学习canal2&#39;);
Salin selepas log masuk

Sisipan, kemas kini dan pemadaman masing-masing diproses

Mulakan program ujian dahulu:

Bagaimana SpringBoot menyepadukan kaedah Canal

<🎜. >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());
            }

        }
    }
Salin selepas log masuk

dan maklumat akan dicetak:

Bagaimana SpringBoot menyepadukan kaedah Canal

Ini sangat boleh dibaca.

Jadual utama melakukan operasi pengubahsuaian.

insert into springboot.user(id,name,age,sex,description) 
values(4,&#39;canal添加用户4&#39;,25,&#39;男&#39;,&#39;学习canal4&#39;);
Salin selepas log masuk

Apabila mengemas kini, jika setiap medan adalah sama seperti sebelumnya, tiada penggunaan log akan dijana.

Bagaimana SpringBoot menyepadukan kaedah Canal

Jadual utama melaksanakan operasi pemadaman:

update springboot.user set name=&#39;开开心心&#39;,age=26,description=&#39;岳泽霖&#39; where id =4;
Salin selepas log masuk

Bagaimana SpringBoot menyepadukan kaedah Canal

Pemerolehan di atas semuanya diperolehi satu data pada satu masa . Kecekapan agak rendah

Mendapatkan berbilang keping data pada satu masa

delete from springboot.user where id =4;
Salin selepas log masuk

Titik pengubahsuaian:

/**
     * 一次性获取多条数据。
     * 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());
            }
        }
    }
Salin selepas log masuk

.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);
Salin selepas log masuk

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());
            }
        }
    }
Salin selepas log masuk

Jika ia ditukar kepada pengesahan manual, Bagaimana SpringBoot menyepadukan kaedah Canal

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!

Kenyataan Laman Web ini
Kandungan artikel ini disumbangkan secara sukarela oleh netizen, dan hak cipta adalah milik pengarang asal. Laman web ini tidak memikul tanggungjawab undang-undang yang sepadan. Jika anda menemui sebarang kandungan yang disyaki plagiarisme atau pelanggaran, sila hubungi admin@php.cn

Tag artikel panas

Notepad++7.3.1

Notepad++7.3.1

Editor kod yang mudah digunakan dan percuma

SublimeText3 versi Cina

SublimeText3 versi Cina

Versi Cina, sangat mudah digunakan

Hantar Studio 13.0.1

Hantar Studio 13.0.1

Persekitaran pembangunan bersepadu PHP yang berkuasa

Dreamweaver CS6

Dreamweaver CS6

Alat pembangunan web visual

SublimeText3 versi Mac

SublimeText3 versi Mac

Perisian penyuntingan kod peringkat Tuhan (SublimeText3)

Bagaimana Springboot menyepadukan Jasypt untuk melaksanakan penyulitan fail konfigurasi Bagaimana Springboot menyepadukan Jasypt untuk melaksanakan penyulitan fail konfigurasi Jun 01, 2023 am 08:55 AM

Bagaimana Springboot menyepadukan Jasypt untuk melaksanakan penyulitan fail konfigurasi

Cara menggunakan Redis untuk melaksanakan kunci teragih dalam SpringBoot Cara menggunakan Redis untuk melaksanakan kunci teragih dalam SpringBoot Jun 03, 2023 am 08:16 AM

Cara menggunakan Redis untuk melaksanakan kunci teragih dalam SpringBoot

Bagaimana SpringBoot menyepadukan Redisson untuk melaksanakan baris gilir kelewatan Bagaimana SpringBoot menyepadukan Redisson untuk melaksanakan baris gilir kelewatan May 30, 2023 pm 02:40 PM

Bagaimana SpringBoot menyepadukan Redisson untuk melaksanakan baris gilir kelewatan

Analisis perbandingan dan perbezaan antara SpringBoot dan SpringMVC Analisis perbandingan dan perbezaan antara SpringBoot dan SpringMVC Dec 29, 2023 am 11:02 AM

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 untuk menyelesaikan masalah bahawa springboot tidak boleh mengakses fail selepas membacanya ke dalam pakej balang Jun 03, 2023 pm 04:38 PM

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 SpringBoot menyesuaikan Redis untuk melaksanakan penyirian cache Jun 03, 2023 am 11:32 AM

Bagaimana SpringBoot menyesuaikan Redis untuk melaksanakan penyirian cache

Bagaimana untuk mendapatkan nilai dalam application.yml dalam springboot Bagaimana untuk mendapatkan nilai dalam application.yml dalam springboot Jun 03, 2023 pm 06:43 PM

Bagaimana untuk mendapatkan nilai dalam application.yml dalam springboot

Bagaimana untuk melaksanakan Springboot+Mybatis-plus tanpa menggunakan pernyataan SQL untuk menambah berbilang jadual Bagaimana untuk melaksanakan Springboot+Mybatis-plus tanpa menggunakan pernyataan SQL untuk menambah berbilang jadual Jun 02, 2023 am 11:07 AM

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

See all articles