Mysql启动binlog以及canal监听还有使用的意义

忘是亡心i 2021-10-12 21:00 252阅读 0赞

**目录**

启动binlog

Canal插件

服务端

修改配置

启动

客户端

效果

新增

更新

删除

意义

--------------------

# 启动binlog #

在mysql my.ini进行修改

# Server Id.
    server-id=1
    log-bin=mysql-bin
    binlog-format=ROW

log-bin名字随意起

有时ini配置文件一直找不到？可以去服务里面进行搜索mysql服务，然后查看属性

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70][]

# Canal插件 #

## 服务端 ##

[https://github.com/alibaba/canal/releases][https_github.com_alibaba_canal_releases]进行下载，不过很慢

[https://github.com/alibaba/canal][https_github.com_alibaba_canal] 介绍，伪装成mysql slave进行同步通讯获取binlog信息

[https://github.com/alibaba/canal/wiki/QuickStart][https_github.com_alibaba_canal_wiki_QuickStart]启动项

可以下载我旧版本的（服务端）

链接：[https://pan.baidu.com/s/14Wcw2x2WFBF1fBRMXIHH-g ][https_pan.baidu.com_s_14Wcw2x2WFBF1fBRMXIHH-g]  
提取码：ilb3

### 修改配置 ###

instance.properties（conf\\example目录下）

#################################################
    ## mysql serverId , v1.0.26+ will autoGen 
    # canal.instance.mysql.slaveId=0
    
    # enable gtid use true/false
    canal.instance.gtidon=false
    
    # position info
    canal.instance.master.address=localhost:3306
    canal.instance.master.journal.name=
    canal.instance.master.position=
    canal.instance.master.timestamp=
    canal.instance.master.gtid=
    
    # rds oss binlog
    canal.instance.rds.accesskey=
    canal.instance.rds.secretkey=
    canal.instance.rds.instanceId=
    
    # table meta tsdb info
    canal.instance.tsdb.enable=true
    #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
    #canal.instance.tsdb.dbUsername=canal
    #canal.instance.tsdb.dbPassword=canal
    
    #canal.instance.standby.address =
    #canal.instance.standby.journal.name =
    #canal.instance.standby.position = 
    #canal.instance.standby.timestamp =
    #canal.instance.standby.gtid=
    
    # username/password
    canal.instance.dbUsername=root
    canal.instance.dbPassword=123456
    canal.instance.connectionCharset=UTF-8
    
    canal.instance.defaultDatabaseName = test
    
    # table regex
    canal.instance.filter.regex=.*\\..*
    # table black regex
    canal.instance.filter.black.regex=
    #################################################

使用数据库的账号密码（root 123456那里）

新建一个用户就不用改账号密码，如下：

CREATE USER canal IDENTIFIED BY 'canal';  
    GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
    -- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
    FLUSH PRIVILEGES;

canal.instance.defaultDatabaseName是监听的数据库名称

### 启动 ###

在bin目录下启动，可以去logs下看example还有canal的日志，没有错误就是成功，我之前是报3306连接不了。

## 客户端 ##

pom.xml

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

执行的类

package com.example.demo;
    
    import com.alibaba.otter.canal.client.CanalConnector;
    import com.alibaba.otter.canal.client.CanalConnectors;
    import com.alibaba.otter.canal.common.utils.AddressUtils;
    import com.alibaba.otter.canal.protocol.CanalEntry;
    import com.alibaba.otter.canal.protocol.Message;
    
    import java.net.InetSocketAddress;
    import java.util.List;
    
    /**
     * A Camel Application
     */
    public class MainApp {
    
        /**
         * A main() so we can easily run these routing rules in our IDE
         */
        public static void main(String... args) throws Exception {
    
            // 创建链接
            CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),
                    11111), "example", "", "");
            int batchSize = 1000;
            int emptyCount = 0;
            try {
                connector.connect();
                connector.subscribe(".*\\..*");
                connector.rollback();
                int totalEmptyCount = 12000;
                while (emptyCount < totalEmptyCount) {
                    Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                    long batchId = message.getId();
                    int size = message.getEntries().size();
                    if (batchId == -1 || size == 0) {
                        emptyCount++;
                        System.out.println("empty count : " + emptyCount);
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                        }
                    } else {
                        emptyCount = 0;
                        // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
                        printEntry(message.getEntries());
                    }
    
                    connector.ack(batchId); // 提交确认
                    // connector.rollback(batchId); // 处理失败, 回滚数据
                }
    
                System.out.println("empty too many times, exit");
            } finally {
                connector.disconnect();
            }
        }
    
        private static void printEntry(List<CanalEntry.Entry> entrys) {
            for (CanalEntry.Entry entry : entrys) {
                if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry
                        .EntryType
                        .TRANSACTIONEND) {
                    continue;
                }
    
                CanalEntry.RowChange rowChage = null;
                try {
                    rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) {
                    throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                            e);
                }
    
                CanalEntry.EventType eventType = rowChage.getEventType();
                System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
                        entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                        entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                        eventType));
    
                for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                    if (eventType == CanalEntry.EventType.DELETE) {
                        printColumn(rowData.getBeforeColumnsList());
                    } else if (eventType == CanalEntry.EventType.INSERT) {
                        printColumn(rowData.getAfterColumnsList());
                    } else {
                        System.out.println("-------> before");
                        printColumn(rowData.getBeforeColumnsList());
                        System.out.println("-------> after");
                        printColumn(rowData.getAfterColumnsList());
                    }
                }
            }
        }
    
        private static void printColumn(List<CanalEntry.Column> columns) {
            for (CanalEntry.Column column : columns) {
                System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
            }
        }
    }

## 效果 ##

像上面我配置的是监听test数据库，我们对test数据库里面某个表进行操作

### 新增 ###

================> binlog[mysql-bin.000004:1983] , name[test,test] , eventType : INSERT
    uid : 3735    update=true
    iid : 100    update=true
    score : 5.0    update=true
    ts :     update=true
    id : 387595    update=true

name 数据库 数据表 新增类型

下面是各种字段 update是 是否修改了数据

### 更新 ###

================> binlog[mysql-bin.000004:1744] , name[test,test] , eventType : UPDATE
    -------> before
    uid : 3735    update=false
    iid : 39    update=false
    score : 5.9    update=false
    ts : 2019-08-02 11:10:27    update=false
    id : 387594    update=false
    -------> after
    uid : 3735    update=false
    iid : 40    update=true
    score : 5.9    update=false
    ts : 2019-08-02 11:10:27    update=false
    id : 387594    update=false

之前的数据也有，修改后的数据也有.

update有所变化，如果是修改的为true

### 删除 ###

同新增一样

# 意义 #

binlog我的第一个认识是进行mysql的主从同步，这里是监听binlog的变化，可以进行实时数据变化的更新。

这样的好处？

如果你要做实时的，我们之前公司是进行上报数据，就是每次请求后端一个接口，然后上报另一个接口进行统计。这对带宽，以及多次请求

主要用于实时数据的统计 做个定时任务去一直统计也行，但是不是实时的，第二计算量可能会蛮大，占用cpu时间长等等。

监听完执行相应的业务，canal以及可以将变化传到mq或者通过spark,flink进行消费储存到hbase中，或者统计so on...

[参考文章][Link 1]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70]: /images/20211013/6710744e23df47ec92f75abd02b0b0bc.png
[https_github.com_alibaba_canal_releases]: https://github.com/alibaba/canal/releases
[https_github.com_alibaba_canal]: https://github.com/alibaba/canal
[https_github.com_alibaba_canal_wiki_QuickStart]: https://github.com/alibaba/canal/wiki/QuickStart
[https_pan.baidu.com_s_14Wcw2x2WFBF1fBRMXIHH-g]: https://pan.baidu.com/s/14Wcw2x2WFBF1fBRMXIHH-g%C2%A0
[Link 1]: https://blog.csdn.net/tianyaleixiaowu/article/details/79653829