canal学习笔记（应用案例）

布满荆棘的人生 2023-07-22 09:02 3阅读 0赞

# canal学习笔记（应用案例） #

### 文章目录 ###

*  canal学习笔记（应用案例）
 *   *  一. 场景
     *  二.建立项目
     *  三.场景模拟
     *  canal进阶内容

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h6Xzk0MDYxMQ_size_16_color_FFFFFF_t_70]  
现在MySQL有了，canal服务也有了，就差canal的客户端了。客户端拿到数据可以进行业务处理，这里以java项目为案例。

## 一. 场景 ##

以异构数据的同步为例，假设要将MySQL的数据存入redis和es中。以前的方法都是手动在MySQL更新完成后新增redis和es，现在利用canal监听MySQL的binlog后，再去处理redis和es。这里不考虑后续是采用同步或是异步消息的形式，重点关心canal客户端的建立。

## 二.建立项目 ##

1.  引入canal客户端jar包：

<dependency>
                <groupId>com.alibaba.otter</groupId>
                <artifactId>canal.client</artifactId>
                <version>1.1.4</version>
            </dependency>

1.  编写cana监听客户端  
    测试代码如下：

package com.haozz.canal_client_demo.canal;
    
    import com.alibaba.otter.canal.client.CanalConnector;
    import com.alibaba.otter.canal.client.CanalConnectors;
    import com.alibaba.otter.canal.protocol.CanalEntry;
    import com.alibaba.otter.canal.protocol.Message;
    import com.google.protobuf.InvalidProtocolBufferException;
    
    import java.net.InetSocketAddress;
    import java.util.List;
    
    /** * @author: haozz * @date: 2020/4/4 23:46 */
    public class CanalClient { 
    
        /** * canal_server地址，这里是我本地的虚拟机 */
        private static String SERVER_ADDRESS = "192.168.124.26";
    
        private static Integer PORT = 11111;
    
        /** * 目的地 * canal server中的数据实际上是放在内部的消息队列中，这里需要指定队列的名称 */
        private static String DESTINATION = "example";
    
        private static String USERNAME = "";
    
        private static String PASSWORD = "";
    
    
        public static void main(String[] args) { 
            //建立简介
            CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress(SERVER_ADDRESS, PORT), DESTINATION, USERNAME, PASSWORD);
            canalConnector.connect();
    
            //订阅所有库所有表下的数据变动
            canalConnector.subscribe(".*\\..*");
            canalConnector.rollback();
    
    
            for (; ; ) { 
    
                // 获取指定数量的数据，但不做确认
                // 不做确认的含义是，canal server不做标记，下次取还能取到
                Message message = canalConnector.getWithoutAck(100);
                long messageId = message.getId();
                if (messageId != -1) { 
                    System.out.println(message.getEntries());
                    System.out.println("messageId -> " + messageId);
                    printEntity(message.getEntries());
                }
            }
        }
    
    
        public static void printEntity(List<CanalEntry.Entry> entries) { 
            for (CanalEntry.Entry entry : entries) { 
                if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) { 
                    continue;
                }
    
                try { 
                    CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                    for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { 
                        String tableName = entry.getHeader().getTableName();
    // System.out.println(rowChange.getEventType());
                        switch (rowChange.getEventType()) { 
                            case INSERT:
                                // do something ...
                                System.out.println("this is INSERT for " + tableName + "-> " + rowData.toString());
                                break;
                            case DELETE:
                                // do something ...
                                System.out.println("this is DELETE for " + tableName + "-> " + rowData.toString());
                                break;
                            case UPDATE:
                                // do something ...
                                System.out.println("this is UPDATE for " + tableName + "-> " + rowData.toString());
                                break;
                            default:
                                break;
                        }
                    }
                } catch (InvalidProtocolBufferException e) { 
                    e.printStackTrace();
                }
            }
        }
    
    
    }

## 三.场景模拟 ##

测试效果，当在Navicat中分别执行以下sql时，

INSERT INTO user ( `name`, `age` ) VALUES	( 'robin', 30 );
    update user set name = 'nami' where id = 5;

上面测试类控制台有如下打印：

messageId -> 10
    this is INSERT for user-> afterColumns { 
      index: 0
      sqlType: -5
      name: "id"
      isKey: true
      updated: true
      isNull: false
      value: "5"
      mysqlType: "bigint(20)"
    }
    afterColumns { 
      index: 1
      sqlType: 12
      name: "name"
      isKey: false
      updated: true
      isNull: false
      value: "robin"
      mysqlType: "varchar(255)"
    }
    afterColumns { 
      index: 2
      sqlType: 4
      name: "age"
      isKey: false
      updated: true
      isNull: false
      value: "30"
      mysqlType: "int(11)"
    }
    afterColumns { 
      index: 3
      sqlType: 12
      name: "address"
      isKey: false
      updated: true
      isNull: true
      mysqlType: "varchar(255)"
    }
    ========================================================================
    messageId -> 11
    this is UPDATE for user-> beforeColumns { 
      index: 0
      sqlType: -5
      name: "id"
      isKey: true
      updated: false
      isNull: false
      value: "5"
      mysqlType: "bigint(20)"
    }
    beforeColumns { 
      index: 1
      sqlType: 12
      name: "name"
      isKey: false
      updated: false
      isNull: false
      value: "robin"
      mysqlType: "varchar(255)"
    }
    beforeColumns { 
      index: 2
      sqlType: 4
      name: "age"
      isKey: false
      updated: false
      isNull: false
      value: "30"
      mysqlType: "int(11)"
    }
    beforeColumns { 
      index: 3
      sqlType: 12
      name: "address"
      isKey: false
      updated: false
      isNull: true
      mysqlType: "varchar(255)"
    }
    afterColumns { 
      index: 0
      sqlType: -5
      name: "id"
      isKey: true
      updated: false
      isNull: false
      value: "5"
      mysqlType: "bigint(20)"
    }
    afterColumns { 
      index: 1
      sqlType: 12
      name: "name"
      isKey: false
      updated: true
      isNull: false
      value: "nami"
      mysqlType: "varchar(255)"
    }
    afterColumns { 
      index: 2
      sqlType: 4
      name: "age"
      isKey: false
      updated: false
      isNull: false
      value: "30"
      mysqlType: "int(11)"
    }
    afterColumns { 
      index: 3
      sqlType: 12
      name: "address"
      isKey: false
      updated: false
      isNull: true
      mysqlType: "varchar(255)"
    }
    ========================================================================
    messageId -> 12
    this is DELETE for user-> beforeColumns { 
      index: 0
      sqlType: -5
      name: "id"
      isKey: true
      updated: false
      isNull: false
      value: "5"
      mysqlType: "bigint(20)"
    }
    beforeColumns { 
      index: 1
      sqlType: 12
      name: "name"
      isKey: false
      updated: false
      isNull: false
      value: "nami"
      mysqlType: "varchar(255)"
    }
    beforeColumns { 
      index: 2
      sqlType: 4
      name: "age"
      isKey: false
      updated: false
      isNull: false
      value: "30"
      mysqlType: "int(11)"
    }
    beforeColumns { 
      index: 3
      sqlType: 12
      name: "address"
      isKey: false
      updated: false
      isNull: true
      mysqlType: "varchar(255)"
    }

可以看到，canal不仅监听到了MySQL的数据变化，还可以明确的清楚每一个字段的数据。一般情况下后续需要自定义数据转换类，用Java反射拿到具体的业务对象，然后进行其他业务处理。

## canal进阶内容 ##

在实际开发中，如果只有一台canal机器作为server，当该机器挂掉后，整个服务就会终止，此时就需要引入集群部署方式…

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h6Xzk0MDYxMQ_size_16_color_FFFFFF_t_70]: /images/20230528/4372754ee1944a03b0951be1415d539e.png