package binlog import ( "errors" "eta/eta_bridge/global" "eta/eta_bridge/models/index" "eta/eta_bridge/utils" "fmt" "github.com/go-mysql-org/go-mysql/canal" "github.com/go-mysql-org/go-mysql/mysql" _ "github.com/go-sql-driver/mysql" "math/rand" "strconv" "time" ) //var mysqlHost, mysqlUser, mysqlPwd, mysqlDb string //func init() { // // for _, sqlConfig := range global.CONFIG.Mysql.List { // if sqlConfig.AliasName == `index` { // // 找出用户 // tmpList := strings.Split(sqlConfig.Dsn, ":") // mysqlUser = tmpList[0] // // 找出密码 // tmpList = strings.Split(strings.Join(tmpList[1:], ":"), "@") // lenTmp := len(tmpList) // mysqlPwd = strings.Join(tmpList[:lenTmp-1], "@") // // 找出地址 // tmpList = strings.Split(tmpList[lenTmp-1], "tcp(") // tmpList = strings.Split(tmpList[1], ")") // mysqlHost = tmpList[0] // // // 找出数据库名称 // u, err := url.Parse(tmpList[1]) // if err != nil { // panic(err) // 如果解析失败,处理错误 // } // // // 获取Path部分的内容 // mysqlDb = u.Path[1:] // // } // } // // if mysqlHost == "" { // panic("mysqlHost is empty") // } // // //fmt.Println("HOST:", mysqlHost) // //fmt.Println("user:", mysqlUser) // //fmt.Println("password:", mysqlPwd) // //} func ListenMysql() { var err error defer func() { if err != nil { fmt.Println("数据库监听服务异常,err:", err) } }() if global.CONFIG.Mysql.Binlog.Host == "" { panic("mysqlHost is empty") //err = errors.New("mysqlHost is empty") //return } if global.CONFIG.Mysql.Binlog.User == "" { panic("user is empty") } if global.CONFIG.Mysql.Binlog.Password == "" { panic("password is empty") } if global.CONFIG.Mysql.Binlog.Db == "" { panic("db is empty") } //includeTableRegex := []string{ // "test_hz_data.edb_info", //} includeTableRegex := []string{ global.CONFIG.Mysql.Binlog.Db + ".edb_info$", global.CONFIG.Mysql.Binlog.Db + ".edb_classify$", global.CONFIG.Mysql.Binlog.Db + ".base_from_mysteel_chemical_index$", global.CONFIG.Mysql.Binlog.Db + ".base_from_smm_index$", global.CONFIG.Mysql.Binlog.Db + ".edb_data*", } // 主从复制的身份id配置,必须全局唯一,如果没有配置的话,那么会随机生成一个 serverId := global.CONFIG.Mysql.Binlog.ServerID if serverId == 0 { serverId = uint32(rand.New(rand.NewSource(time.Now().Unix())).Intn(1000)) + 1001 } cfg := &canal.Config{ // 一个32位无符号整数,用于标识当前 Canal 实例在 MySQL 主从复制体系中的身份。这里使用了一个随机数生成器确保每次启动时分配的 ServerID 是唯一的(在1001到1099之间)。在实际生产环境中,你需要手动指定一个全局唯一的 ServerID。 ServerID: serverId, // 指定 Canal 要连接的数据库类型,默认为 "mysql",表明这是一个 MySQL 数据库。 Flavor: "mysql", // 设置 MySQL 服务器地址(主机名或 IP 地址)和端口,例如 "127.0.0.1:3306"。 Addr: global.CONFIG.Mysql.Binlog.Host, User: global.CONFIG.Mysql.Binlog.User, Password: global.CONFIG.Mysql.Binlog.Password, // 如果设置为 true,Canal 将以原始二进制格式获取 binlog,否则将以解析后的 SQL 语句形式提供。 //RawModeEnabled: false, // 是否启用半同步复制。当设置为 true 时,MySQL 主库在事务提交后会等待至少一个从库确认已接收并写入 binlog 才返回成功,提高了数据一致性。 SemiSyncEnabled: false, // 是否将 MySQL 中的 decimal 类型字段解析为 Go 的 Decimal 类型,而不是 float 或者 string。如果业务中有精确小数计算的需求,应开启此选项以避免精度丢失问题。 UseDecimal: true, // 用于控制初始数据导出的相关配置,在 Canal 启动时是否需要全量同步表数据。 //Dump: dumpConf, // 正则表达式字符串,用于定义 Canal 应该监听哪些表的 binlog 事件。只有名称匹配该正则表达式的表才会被 Canal 同步处理。 IncludeTableRegex: includeTableRegex, } // 校验mysql binlog format,目前仅支持row格式 { binlogFormat, tmpErr := index.GetBinlogFormat() if tmpErr != nil { err = tmpErr return } if binlogFormat.Value != "ROW" { panic("mysql binlog format is not ROW") return } } // 获取上一次启动时的binlog文件名称和位置 fileName, position, err := getBinlogNamePosition() if err != nil { return } // 修改记录本次启动时的binlog文件名称和位置 modifyBinlogNamePosition(fileName, position) // 定时修改binlog文件名称和位置 go timingModifyBinlogNamePosition() c, err := canal.NewCanal(cfg) if err != nil { fmt.Println("err:", err) return } global.FILE_LOG.Debug("记录上一次启动时的fileName:", fileName, ";position:", position) binlogHandler := &MyEventHandler{} binlogHandler.SetBinlogFileName(fileName, position) c.SetEventHandler(binlogHandler) //c.Run() pos := mysql.Position{ Name: fileName, Pos: position, } err = c.RunFrom(pos) } // getBinlogNamePosition // @Description: 获取当前binlog文件名称和位置 // @author: Roc // @datetime 2024-05-17 13:18:19 // @return fileName string // @return position uint32 // @return err error func getBinlogNamePosition() (fileName string, position uint32, err error) { // 优先从redis获取 fileName = global.Rc.GetStr(utils.CACHE_MYSQL_MASTER_FILENAME) position64, err := global.Rc.GetUInt64(utils.CACHE_MYSQL_MASTER_POSITION) if err != nil { if err.Error() != utils.RedisNoKeyErr { panic("mysql binlog position is not found,err:" + err.Error()) return } err = nil } position = uint32(position64) // 如果没有从redis中获取到上次监听到的binlog的文件名称,或者位置为0,则从mysql中获取,则从 MySQL 中获取最新的文件名和位置。 if fileName == `` || position == 0 { // binlog文件名 fileNameKey := index.BinlogFileNameKey fileNameLog, tmpErr := index.GetBusinessSysInteractionLogByKey(fileNameKey) if tmpErr == nil { fileName = fileNameLog.InteractionKey } // binlog位置 positionKey := index.BinlogPositionKey positionLog, tmpErr := index.GetBusinessSysInteractionLogByKey(positionKey) if tmpErr == nil { positionStr := positionLog.InteractionKey positionInt, tmpErr := strconv.Atoi(positionStr) if tmpErr == nil { position = uint32(positionInt) } } } // 如果从表中没有取到数据,则从mysql中获取,则从 MySQL 中获取最新的文件名和位置。 if fileName == `` || position == 0 { item, tmpErr := index.GetShowMaster() if tmpErr != nil { err = tmpErr return } fileName = item.File position = item.Position } return } // timingModifyBinlogNamePosition // @Description: 定时修改binlog文件名称和位置 // @author: Roc // @datetime 2024-05-17 13:08:13 func timingModifyBinlogNamePosition() { for { // 延时30s执行 time.Sleep(30 * time.Second) // 获取最新的binlog文件名称和位置 fileName, position, err := getBinlogNamePosition() if err != nil { return } if fileName != `` && position != 0 { // 修改记录本次启动时的binlog文件名称和位置 modifyBinlogNamePosition(fileName, position) } } } // modifyBinlogNamePosition // @Description: 修改记录本次启动时的binlog文件名称和位置 // @author: Roc // @datetime 2024-05-17 11:32:32 // @param fileName string // @param position uint32 // @return err error func modifyBinlogNamePosition(fileName string, position uint32) { var err error defer func() { if err != nil { global.FILE_LOG.Error("修改binlog文件名称和位置异常,fileName", fileName, ",position:", position, ",err:", err) } }() // fileName 变更 fileNameKey := index.BinlogFileNameKey fileNameLog, err := index.GetBusinessSysInteractionLogByKey(fileNameKey) if err != nil { if !errors.Is(err, utils.ErrNoRow) { return } err = nil fileNameLog = &index.BusinessSysInteractionLog{ //ID: 0, InteractionKey: fileNameKey, InteractionVal: fileName, Remark: "mysql中binlog的filename名称", ModifyTime: time.Now(), CreateTime: time.Now(), } err = fileNameLog.Create() if err != nil { return } } else { fileNameLog.InteractionVal = fileName fileNameLog.ModifyTime = time.Now() err = fileNameLog.Update([]string{"InteractionVal", "ModifyTime"}) if err != nil { return } } // position 变更 positionKey := index.BinlogPositionKey positionLog, err := index.GetBusinessSysInteractionLogByKey(positionKey) if err != nil { if !errors.Is(err, utils.ErrNoRow) { return } err = nil positionLog = &index.BusinessSysInteractionLog{ //ID: 0, InteractionKey: positionKey, InteractionVal: fmt.Sprint(position), Remark: "mysql中binlog的position位置", ModifyTime: time.Now(), CreateTime: time.Now(), } err = positionLog.Create() if err != nil { return } } else { positionLog.InteractionVal = fmt.Sprint(position) positionLog.ModifyTime = time.Now() err = positionLog.Update([]string{"InteractionVal", "ModifyTime"}) if err != nil { return } } return }