eta_api/services/binlog/handler.go

package binlog

import (
	"eta/eta_api/utils"
	"fmt"
	"reflect"
	"time"

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

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

func (h *EdbEventHandler) 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)

	// 每次操作完成后都将当前位置记录到缓存
	utils.Rc.Put(utils.CACHE_MYSQL_DATA_FILENAME, h.fileName, 31*24*time.Hour)
	utils.Rc.Put(utils.CACHE_MYSQL_DATA_POSITION, h.position, 31*24*time.Hour)

	return nil
}

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

	// 旋转binlog日志的时候，需要将当前位置记录到缓存
	utils.Rc.Put(utils.CACHE_MYSQL_DATA_FILENAME, h.fileName, 31*24*time.Hour)
	utils.Rc.Put(utils.CACHE_MYSQL_DATA_POSITION, h.position, 31*24*time.Hour)

	return nil
}

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

func (h *EdbEventHandler) Insert(e *canal.RowsEvent) error {
	// 批量插入的时候，e.Rows的长度会大于0
	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
			}
		}
	}

	return nil
}

func (h *EdbEventHandler) 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
		}

	}

	return nil
}

func (h *EdbEventHandler) Delete(e *canal.RowsEvent) error {
	// 批量删除的时候，e.Rows的长度会大于0
	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
			}
		}
	}

	return nil
}

func (h *EdbEventHandler) 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
		}
	}

	return nil
}

// SetBinlogFileName
// @Description: 设置当前的binlog文件名和位置
// @author: Roc
// @receiver h
// @datetime 2024-02-29 18:09:36
// @param fileName string
// @param position uint32
func (h *EdbEventHandler) SetBinlogFileName(fileName string, position uint32) {
	h.fileName = fileName
	h.position = position

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