Linux宝塔面板使用Canal实现Mysql和Redis数据同步（图文教程）

工作原理流程：
mysql 主库 → binlog（二进制日志） → canal server（伪装成从库） → 解析日志 → 发送数据 → 各种下游系统

# 配置 slaveid 自定义,不等于 mysql 的 server id 即可
canal.instance.mysql.slaveid=10 

# 数据库地址:自己的数据库ip+端口
canal.instance.master.address=127.0.0.1:3306 
 
# 数据库用户名和密码 
canal.instance.dbusername=canal
canal.instance.dbpassword=canal

#代表数据库的编码方式对应到 java 中的编码类型，比如 utf-8，gbk , iso-8859-1
canal.instance.connectioncharset = utf-8
	
# 指定库和表，这里的 .* 表示 canal.instance.master.address 下面的所有数据库
canal.instance.filter.regex=.*\\..*

2023-09-24 12:14:24.280 [main] info  c.a.otter.canal.instance.spring.canalinstancewithspring - start cannalinstance for 1-example 
2023-09-24 12:14:24.309 [main] warn  c.a.o.canal.parse.inbound.mysql.dbsync.logeventconvert - --> init table filter : ^.*\..*$
2023-09-24 12:14:24.309 [main] warn  c.a.o.canal.parse.inbound.mysql.dbsync.logeventconvert - --> init table black filter : ^mysql\.slave_.*$
2023-09-24 12:14:24.320 [main] info  c.a.otter.canal.instance.core.abstractcanalinstance - start successful....
2023-09-24 12:14:24.480 [destination = example , address = /127.0.0.1:3306 , eventparser] warn  c.a.o.c.p.inbound.mysql.rds.rdsbinlogeventparserproxy - ---> begin to find start position, it will be long time for reset or first position
2023-09-24 12:14:24.480 [destination = example , address = /127.0.0.1:3306 , eventparser] warn  c.a.o.c.p.inbound.mysql.rds.rdsbinlogeventparserproxy - prepare to find start position just show master status
2023-09-24 12:14:26.135 [destination = example , address = /127.0.0.1:3306 , eventparser] warn  c.a.o.c.p.inbound.mysql.rds.rdsbinlogeventparserproxy - ---> find start position successfully, entryposition[included=false,journalname=mysql-bin.000045,position=4,serverid=1,gtid=<null>,timestamp=1695526635000] cost : 1634ms , the next step is binlog dump

package com.game.service;

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 java.net.inetsocketaddress;
import java.util.list;

public class canalclient {
    public static void main(string args[]) {
        // 创建链接：换成自己的数据库ip地址
        canalconnector connector = canalconnectors.newsingleconnector(new inetsocketaddress("192.168.0.243",
                11111), "example", "", "");
        int batchsize = 1000;
        int emptycount = 0;
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            int totalemptycount = 120;
            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;
                    printentry(message.getentries());
                }

                connector.ack(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("================&gt; 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("-------&gt; before");
                    printcolumn(rowdata.getbeforecolumnslist());
                    system.out.println("-------&gt; 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());
        }
    }
}