如何使用 Golang 处理 MySQL 的 binlog

大家好，我是 Artem，一名 Golang 开发。我们的团队花费了大量时间训练 MySQL binlog。这里整合一些简单用法，不会放过任何隐藏的陷阱。示例代码将在最后显示。

每次从数据库查询的返回结果中拉取用户信息时，主项目中会有高负载模块。此时使用缓存是一个不错的建议，但是什么时候重置缓存呢？这需要由数据来决定更新时间。

MySQL 的主从复制是一个很棒的设计。而我们的守护进程可以视为一个通过 binlog 获取数据的 slave，binlog 设置成 row 格式。这样就能使用所有的数据库命令，但事务下的命令只有在提交后才会记录。在达到内存的使用限制后（默认为 1GB），会开启另一个文件，每个新文件的名称后都会有一个增量。

更多信息查看 https://mariadb.com/kb/en/library/binary-log/ 或者 https://dev.mysql.com/doc/refman/8.0/en/binary-log.html

本文将分为以下两部分：

1. 如何处理 binlog 中的新数据
2. 如何设置和扩展

part 1. 快速运行

我们可以使用这个库 https://github.com/siddontang/go-mysql/ 来处理 binlog。

连接到一个新的 channel（chanal 是一个库的标签）。我们将使用 binlog 中的 row 格式 https://mariadb.com/kb/en/library/binary-log-formats/ 。

func binLogListener() {
	c, err := getDefaultCanal()
	if err == nil {
		coords, err := c.GetMasterPos()
		if err == nil {
			c.SetEventHandler(&binlogHandler{})
			c.RunFrom(coords)
		}
	}
}
func getDefaultCanal() (*canal.Canal, error) {
	cfg := canal.NewDefaultConfig()
	cfg.Addr = fmt.Sprintf("%s:%d", "127.0.0.1", 3306)
	cfg.User = "root"
	cfg.Password = "root"
	cfg.Flavor = "mysql"
	cfg.Dump.ExecutionPath = ""

	return canal.NewCanal(cfg)
}

现在来进行封装

type binlogHandler struct {
	canal.DummyEventHandler // Dummy handler from external lib
	BinlogParser // Our custom helper
}
func (h *binlogHandler) OnRow(e *canal.RowsEvent) error {return nil}
func (h *binlogHandler) String() string {return "binlogHandler"}

BinlogParser

然后我们可以在 OnRow() 方法中添加一些代码逻辑，让他更好用

func (h *binlogHandler) OnRow(e *canal.RowsEvent) error {
	var n int //starting value
	var k int // step
	switch e.Action {
	case canal.DeleteAction:
		return nil // not covered in example
	case canal.UpdateAction:
		n = 1
		k = 2
	case canal.InsertAction:
		n = 0
		k = 1
	}
	for i := n; i < len(e.Rows); i += k {
		key := e.Table.Schema + "." + e.Table.Name
		switch key {
		case User{}.SchemaName() + "." + User{}.TableName():
			/*
			 Real data parsing
			*/
		}
	}
	return nil
}

这个包装器的主要逻辑是解析接收到的数据。我们可以通过更新的两个条件获取数据（第一条包含初始数据，第二条则是更新数据），同时也支持多行插入和多行更新。在这种情况下，执行 UPDATE 操作时，每次都要使用第二个条件。而执行 INSERT 时，需要操作每一行，为此我们需要使用 n 和 k 变量。

从 binlog 中获取一个模版，逐行加载数据，每个 column 都标明注释：

type User struct {
	Id      int       `gorm:"column:id"`
	Name    string    `gorm:"column:name"`
	Status  string    `gorm:"column:status"`
	Created time.Time `gorm:"column:created"`
}
func (User) TableName() string {
	return "User"
}
func (User) SchemaName() string {
	return "Test"
}

MySQL 的表结构体

CREATE TABLE Test.User(
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(40) NULL ,
status ENUM("active","deleted") DEFAULT "active",
created TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL ON UPDATE CURRENT_TIMESTAMP
)
ENGINE =InnoDB;

现在使用代码的方式实现

user := User{}
h.GetBinLogData(&user, e, i)

最后，我们可以通过新增加的用户变量获取数据。打印出来让它看起来更美观。

if e.Action == canal.UpdateAction {
  oldUser := User{}
  h.GetBinLogData(&oldUser, e, i-1)
  fmt.Printf("User %d is updated from name %s to name %s\n", user.Id, oldUser.Name, user.Name, )
 } else {
  fmt.Printf("User %d is created with name %s\n", user.Id, user.Name, )
}

太好了，代码即将实现， "Hello, binlog world":

func main() {
	go binLogListener()
	// placeholder for your handsome code
	time.Sleep(2 * time.Minute)
	fmt.Print("Thx for watching")
}

新增和更新用户：

INSERT INTO Test.User (`id`,`name`) VALUE (1,"Jack");
UPDATE Test.User SET name="Jonh" WHERE id=1;

结果展示：

User 1 is created with name Jack
User 1 name changed from Jack to Jonh

这段代码通过 binlog 来解析新增的 row，并通过数据表获取我们需要的数据，在结构体中解析数据并输出结果。我没有介绍所有的数据解析器（BinlogParser），这其中还隐藏了一些 hydration 逻辑模型。

part 2. 正如 cobb 所说，我们需要更加深入了解

解析器的隐藏部分是基于反射，可以使用下面这种方式来进行 hydration 模型。

h.GetBinLogData(&user, e, i)

使用一些简单的数据类型来处理

bool
int
float64
string
time.Time

也可以通过 JSON 来解析结构体。

如果你需要更多的数据类型 , 或者你只是想知道 binlog 是如何进行解析工作的 , 最好的办法是自己扩展解析类型。

下面是一个 int 类型的实例 :

type User struct {
 Id      int       `gorm:"column:id"`
}

我们可以通过反射来获取类型名称。parseTagSetting 方法可以使注释更便于使用：

element := User{} //In common cases we have interface, but here we will start with model
v := reflect.ValueOf(element)
s := reflect.Indirect(v)
t := s.Type()
num := t.NumField()
parsedTag := parseTagSetting(t.Field(k).Tag)
if columnName, ok = parsedTag["COLUMN"]; !ok || columnName == "COLUMN" {
	continue
 }
for k := 0; k < num; k++ {
	name := s.Field(k).Type().Name()
	switch name {
		case "int":
		// here we deal with an incoming row
	}
}

获取了类型名称的同时也可以通过反射来设置它的值

func (v Value) SetInt(x int64) {//...

解析注释帮助器（从 Gorm 库获取）

func parseTagSetting(tags reflect.StructTag) map[string]string {
	setting := map[string]string{}
	for _, str := range []string{tags.Get("sql"), tags.Get("gorm")} {
		tags := strings.Split(str, ";")
		for _, value := range tags {
			v := strings.Split(value, ":")
			k := strings.TrimSpace(strings.ToUpper(v[0]))
			if len(v) >= 2 {
				setting[k] = strings.Join(v[1:], ":")
			} else {
				setting[k] = k
			}
		}
	}
	return setting
}

解析器中有 int64 类型，我们可以创建一个将 int64 转换为 row 类型的方法：

func (m *BinlogParser) intHelper(e *canal.RowsEvent, n int, columnName string) int64 {
	columnId := m.getBinlogIdByName(e, columnName)
	if e.Table.Columns[columnId].Type != schema.TYPE_NUMBER {
		return 0
	}
	switch e.Rows[n][columnId].(type) {
	case int8:
		return int64(e.Rows[n][columnId].(int8))
	case int32:
		return int64(e.Rows[n][columnId].(int32))
	case int64:
		return e.Rows[n][columnId].(int64)
	case int:
		return int64(e.Rows[n][columnId].(int))
	case uint8:
		return int64(e.Rows[n][columnId].(uint8))
	case uint16:
		return int64(e.Rows[n][columnId].(uint16))
	case uint32:
		return int64(e.Rows[n][columnId].(uint32))
	case uint64:
		return int64(e.Rows[n][columnId].(uint64))
	case uint:
		return int64(e.Rows[n][columnId].(uint))
	}
	return 0
}

除了 getBinlogIdByName() 方法，所有东西看起来都是合理的。

需要使用 trivial 帮助器来处理 column 名而不是它的 id，这样可以：

使用 Gorm 注释来处理字段名：

在开头和中间添加字段名时不需要额外修改：

使用字段名处理比 column3 更方便。

最后，我们加入以下处理：

s.Field(k).SetInt(m.intHelper(e, n, columnName))

还有两个例子

ENUM: 我们将获取的值作为索引——所以“ active ”状态会被设置为 0。同样的，我们也需要用 enum 字符串表示，而不是 id，这些可以从字段介绍中获取。重要提示，值中的 1 描述的是 0 值索引字段，数组的值是从 0 开始的。

Enum 的解析如下：

func (m *BinlogParser) stringHelper(e *canal.RowsEvent, n int, columnName string) string {
	columnId := m.getBinlogIdByName(e, columnName)
	if e.Table.Columns[columnId].Type == schema.TYPE_ENUM {
		values := e.Table.Columns[columnId].EnumValues //fields value
		if len(values) == 0 || e.Rows[n][columnId] == nil {
			return ""
		}
		return values[e.Rows[n][columnId].(int64)-1] // first id in result is zero one in values
	}
}

存储 JSON

这难道不是个好主意吗？ JSON 是 MySQL 引擎侧的字符串，我们可以将序列化的数据指向解析器。为此，可以添加一个自定义 Gorm 注释——“ fromJson ”，以下是不同数据之间的例子：

type JsonData struct {
	Int        int               `gorm:"column:int"`
	StructData TestData          `gorm:"column:struct_data;fromJson"`
	MapData    map[string]string `gorm:"column:map_data;fromJson"`
	SliceData  []int             `gorm:"column:slice_data;fromJson"`
}
type TestData struct {
	Test string `json:"test"`
	Int  int    `json:"int"`
}

虽然可以创造很多条件来实现，但是新增字段会损坏它。上 Stack Overflow 寻找答案的结果可能是，“如何从未知的 JSON 结构体解析 ? ” “ 不知道你为什么需要这样，但你可以试试 ...”

将结构体转换为接口可以实现：

if _, ok := parsedTag["FROMJSON"]; ok {
	newObject := reflect.New(s.Field(k).Type()).Interface()
	json := m.stringHelper(e, n, columnName)
	jsoniter.Unmarshal([]byte(json), &newObject)
	s.Field(k).Set(reflect.ValueOf(newObject).Elem().Convert(s.Field(k).Type()))
}

如果还有问题、更正或者建议，欢迎提出。此外，需要校对的地方可以在这里提出： https://github.com/JackShadow/go-binlog-example/blob/master/src/parser_test.go

代码示例： https://github.com/JackShadow/go-binlog-example

特别感谢： Freadm Project.