123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271 |
- package binlog
- import (
- "eta/eta_api/models/binlog"
- "eta/eta_api/utils"
- "fmt"
- "math/rand"
- "strconv"
- "time"
- "github.com/go-mysql-org/go-mysql/canal"
- "github.com/go-mysql-org/go-mysql/mysql"
- _ "github.com/go-sql-driver/mysql"
- )
- func ListenMysql() {
- var err error
- defer func() {
- if err != nil {
- fmt.Println("数据库监听服务异常,err:", err)
- }
- }()
- if utils.MYSQL_DATA_BINLOG_URL == "" {
- panic("mysql url is empty")
- }
- if utils.MYSQL_DATA_BINLOG_USER == "" {
- panic("mysql user is empty")
- }
- if utils.MYSQL_DATA_BINLOG_PWD == "" {
- panic("mysql password is empty")
- }
- if utils.MYSQL_DATA_BINLOG_DB == "" {
- panic("mysql db is empty")
- }
- includeTableRegex := []string{
- utils.MYSQL_DATA_BINLOG_DB + ".edb_info$",
-
-
-
-
- }
-
- var serverId uint32
- if utils.MYSQL_DATA_BINLOG_SERVER_ID != "" {
- id, _ := strconv.ParseUint(utils.MYSQL_DATA_BINLOG_SERVER_ID, 10, 32)
- serverId = uint32(id)
- }
- if serverId == 0 {
- serverId = uint32(rand.New(rand.NewSource(time.Now().Unix())).Intn(1000)) + 1001
- }
- cfg := &canal.Config{
-
- ServerID: serverId,
-
- Flavor: "mysql",
-
- Addr: utils.MYSQL_DATA_BINLOG_URL,
- User: utils.MYSQL_DATA_BINLOG_USER,
- Password: utils.MYSQL_DATA_BINLOG_PWD,
-
-
-
- SemiSyncEnabled: false,
-
- UseDecimal: true,
-
-
-
- IncludeTableRegex: includeTableRegex,
- }
-
- {
- binlogFormat, tmpErr := binlog.GetBinlogFormat()
- if tmpErr != nil {
- err = tmpErr
- return
- }
- if binlogFormat.Value != "ROW" {
- panic("mysql binlog format is not ROW")
- }
- }
-
- 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
- }
- utils.FileLog.Debug("记录上一次启动时的fileName:", fileName, ";position:", position)
- binlogHandler := &EdbEventHandler{}
- binlogHandler.SetBinlogFileName(fileName, position)
- c.SetEventHandler(binlogHandler)
-
-
- go binlogHandler.SyncToRedis()
- pos := mysql.Position{
- Name: fileName,
- Pos: position,
- }
- err = c.RunFrom(pos)
- }
- func getBinlogNamePosition() (fileName string, position uint32, err error) {
-
- fileName = utils.Rc.GetStr(utils.CACHE_MYSQL_DATA_FILENAME)
- position64, err := utils.Rc.GetUInt64(utils.CACHE_MYSQL_DATA_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 := binlog.BinlogFileNameKey
- fileNameLog, tmpErr := binlog.GetBusinessSysInteractionLogByKey(fileNameKey)
- if tmpErr == nil {
- fileName = fileNameLog.InteractionKey
- }
-
- positionKey := binlog.BinlogPositionKey
- positionLog, tmpErr := binlog.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 := binlog.GetShowMaster()
- if tmpErr != nil {
- err = tmpErr
- return
- }
- fileName = item.File
- position = item.Position
- }
- return
- }
- func modifyBinlogNamePosition(fileName string, position uint32) {
- var err error
- defer func() {
- if err != nil {
- utils.FileLog.Error("修改binlog文件名称和位置异常,fileName", fileName, ",position:", position, ",err:", err)
- }
- }()
-
- fileNameKey := binlog.BinlogFileNameKey
- fileNameLog, err := binlog.GetBusinessSysInteractionLogByKey(fileNameKey)
- if err != nil {
- if err.Error() != utils.ErrNoRow() {
- return
- }
- err = nil
- fileNameLog = &binlog.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 := binlog.BinlogPositionKey
- positionLog, err := binlog.GetBusinessSysInteractionLogByKey(positionKey)
- if err != nil {
- if err.Error() != utils.ErrNoRow() {
- return
- }
- err = nil
- positionLog = &binlog.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
- }
- func timingModifyBinlogNamePosition() {
- for {
-
- time.Sleep(30 * time.Second)
-
- fileName, position, err := getBinlogNamePosition()
- if err != nil {
- return
- }
- if fileName != `` && position != 0 {
-
- modifyBinlogNamePosition(fileName, position)
- }
- }
- }
|