eta_bridge/services/binlog/handler.go

package binlog

import (
	"encoding/json"
	"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-mysql-org/go-mysql/replication"
	"github.com/pingcap/errors"
	"reflect"
	"time"
)

type MyEventHandler struct {
	canal.DummyEventHandler
	fileName string
	position uint32
}

func (h *MyEventHandler) OnRow(e *canal.RowsEvent) (err error) {

	switch e.Action {
	case canal.InsertAction:
		err = h.Insert(e)
	case canal.UpdateAction:
		err = h.Update(e)
	case canal.DeleteAction:
		err = h.Delete(e)
	default:
		return errors.New("操作异常")
	}

	fmt.Println("fileName:", h.fileName, ";position:", h.position)

	global.Rc.Put(utils.CACHE_MYSQL_MASTER_FILENAME, h.fileName, 31*24*time.Hour)
	global.Rc.Put(utils.CACHE_MYSQL_MASTER_POSITION, h.position, 31*24*time.Hour)

	return nil
}

func (h *MyEventHandler) OnPosSynced(header *replication.EventHeader, p mysql.Position, set mysql.GTIDSet, f bool) error {
	h.fileName = p.Name
	h.position = p.Pos

	global.Rc.Put(utils.CACHE_MYSQL_MASTER_FILENAME, h.fileName, 31*24*time.Hour)
	global.Rc.Put(utils.CACHE_MYSQL_MASTER_POSITION, h.position, 31*24*time.Hour)

	return nil
}

func (h *MyEventHandler) String() string {
	return "MyEventHandler"
}

func (h *MyEventHandler) Insert(e *canal.RowsEvent) error {
	fmt.Println(e.Header.ServerID, ";", e.Table.Schema, ".", e.Table.Name)

	for _, row := range e.Rows { // 遍历当前插入的数据列表（存在批量插入的情况，所以是list）
		logData := make(map[string]interface{})
		dataLen := len(row)
		for i, v := range e.Table.Columns {
			if i < dataLen {
				tmpData := row[i]
				if tmpData != nil && reflect.TypeOf(tmpData).Kind() == reflect.Slice {
					tmpOld := tmpData.([]byte)
					tmpData = string(tmpOld)
				}
				logData[v.Name] = tmpData
			}
		}
		dataByte, _ := json.Marshal(logData)
		log(e.Table.Schema, e.Table.Name, e.Action, ``, string(dataByte))
	}

	return nil
}

func (h *MyEventHandler) Update(e *canal.RowsEvent) error {
	if len(e.Rows) != 2 {
		fmt.Println("更新数据异常，没有原始数据和新数据：", e.Rows)
		return nil
	}


	logOldData := make(map[string]interface{})
	logNewData := make(map[string]interface{})

	oldDataLen := len(e.Rows[0])
	newDataLen := len(e.Rows[0])
	for i, v := range e.Table.Columns {

		if i < oldDataLen {
			oldData := e.Rows[0][i]
			if oldData != nil && reflect.TypeOf(oldData).Kind() == reflect.Slice {
				tmpOld := oldData.([]byte)
				oldData = string(tmpOld)
			}
			logOldData[v.Name] = oldData
		}
		if i < newDataLen {
			newData := e.Rows[1][i]
			if newData != nil && reflect.TypeOf(newData).Kind() == reflect.Slice {
				tmpNew := newData.([]byte)
				newData = string(tmpNew)
			}
			logNewData[v.Name] = newData
		}

	}

	logOldDataByte, _ := json.Marshal(logOldData)
	logNewDataByte, _ := json.Marshal(logNewData)
	log(e.Table.Schema, e.Table.Name, e.Action, string(logOldDataByte), string(logNewDataByte))

	return nil
}

func (h *MyEventHandler) Delete(e *canal.RowsEvent) error {
	fmt.Println(e.Header.ServerID, ";", e.Table.Schema, ".", e.Table.Name)

	for _, row := range e.Rows { // 遍历当前插入的数据列表（存在批量插入的情况，所以是list）
		logData := make(map[string]interface{})
		dataLen := len(row)
		for i, v := range e.Table.Columns {
			if i < dataLen {
				tmpData := row[i]
				if tmpData != nil && reflect.TypeOf(tmpData).Kind() == reflect.Slice {
					tmpOld := tmpData.([]byte)
					tmpData = string(tmpOld)
				}
				logData[v.Name] = tmpData
			}
		}
		dataByte, _ := json.Marshal(logData)
		log(e.Table.Schema, e.Table.Name, e.Action, string(dataByte), ``)
	}

	return nil
}

func (h *MyEventHandler) Delete3(e *canal.RowsEvent) error {
	if len(e.Rows) != 1 {
		fmt.Println("删除数据异常，没有原始数据：", e.Rows)
		return nil
	}
	fmt.Println(e.Header.ServerID, ";", e.Table.Schema, ".", e.Table.Name)

	dataLen := len(e.Rows[0])
	logData := make(map[string]interface{})
	for i, v := range e.Table.Columns {
		if i < dataLen {
			tmpData := e.Rows[0][i]
			if tmpData != nil && reflect.TypeOf(tmpData).Kind() == reflect.Slice {
				tmpOld := tmpData.([]byte)
				tmpData = string(tmpOld)
			}
			logData[v.Name] = tmpData
		}
	}

	dataByte, _ := json.Marshal(logData)
	log(e.Table.Schema, e.Table.Name, e.Action, string(dataByte), ``)

	return nil
}

func (h *MyEventHandler) SetBinlogFileName(fileName string, position uint32) {
	h.fileName = fileName
	h.position = position

	fmt.Println("init fileName:", h.fileName, ";position:", h.position)
}

func log(dbName, tableName, opType, oldData, newData string) {
	item := index.EdbUpdateLog{
		OpDbName:    dbName,
		OpTableName: tableName,
		OpType:      opType,
		OldData:     oldData,
		NewData:     newData,
		CreateTime:  time.Now(),
	}
	err := item.Create()
	if err != nil {
		fmt.Println("log create err:", err.Error())
	}
}