eta_bridge/services/binlog/binlog.go

package binlog

import (
	"errors"
	"eta_gn/eta_bridge/global"
	"eta_gn/eta_bridge/models/index"
	"eta_gn/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"
)



func ListenMysql() {
	var err error
	defer func() {
		if err != nil {
			fmt.Println("数据库监听服务异常,err：", err)
		}
	}()
	if global.CONFIG.Mysql.Binlog.Host == "" {
		panic("mysqlHost is empty")
	}

	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{
		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*",
	}

	serverId := global.CONFIG.Mysql.Binlog.ServerID
	if serverId == 0 {
		serverId = uint32(rand.New(rand.NewSource(time.Now().UnixNano())).Intn(1000)) + 1001
	}
	cfg := &canal.Config{
		ServerID: serverId,
		Flavor: "mysql",
		Addr:     global.CONFIG.Mysql.Binlog.Host,
		User:     global.CONFIG.Mysql.Binlog.User,
		Password: global.CONFIG.Mysql.Binlog.Password,
		SemiSyncEnabled: false,
		UseDecimal: true,
		IncludeTableRegex: includeTableRegex,
	}

	{
		binlogFormat, tmpErr := index.GetBinlogFormat()
		if tmpErr != nil {
			err = tmpErr
			return
		}

		if binlogFormat.Value != "ROW" {
			panic("mysql binlog format is not ROW")
			return
		}
	}

	fileName, position, err := getBinlogNamePosition()
	if err != nil {
		return
	}
	modifyBinlogNamePosition(fileName, position)
	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)

	pos := mysql.Position{
		Name: fileName,
		Pos:  position,
	}
	err = c.RunFrom(pos)
}

func getBinlogNamePosition() (fileName string, position uint32, err error) {
	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)

	if fileName == `` || position == 0 {

		fileNameKey := index.BinlogFileNameKey
		fileNameLog, tmpErr := index.GetBusinessSysInteractionLogByKey(fileNameKey)
		if tmpErr == nil {
			fileName = fileNameLog.InteractionKey
		}

		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)
			}
		}
	}

	if fileName == `` || position == 0 {
		item, tmpErr := index.GetShowMaster()
		if tmpErr != nil {
			err = tmpErr
			return
		}
		fileName = item.File
		position = item.Position
	}

	return
}

func timingModifyBinlogNamePosition() {
	for {
		time.Sleep(30 * time.Second)

		fileName, position, err := getBinlogNamePosition()
		if err != nil {
			return
		}

		if fileName != `` && position != 0 {
			modifyBinlogNamePosition(fileName, position)
		}
	}
}

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)
		}
	}()

	fileNameKey := index.BinlogFileNameKey
	fileNameLog, err := index.GetBusinessSysInteractionLogByKey(fileNameKey)
	if err != nil {
		if !errors.Is(err, utils.ErrNoRow) {
			return
		}
		err = nil
		fileNameLog = &index.BusinessSysInteractionLog{
			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
		}
	}

	positionKey := index.BinlogPositionKey
	positionLog, err := index.GetBusinessSysInteractionLogByKey(positionKey)
	if err != nil {
		if !errors.Is(err, utils.ErrNoRow) {
			return
		}
		err = nil
		positionLog = &index.BusinessSysInteractionLog{
			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
}