https://github.com/go-mysql-org/go-mysql是一个实现了mysql协议和binlog协议的工具库，可以用来实现主从复制（Replication），增量同步（Incremental dumping），客户端（Client），虚拟服务端（Fake server），高可用（Failover），mysql的驱动（database/sql like driver），前面介绍的golang源码分析：mysql同步工具gravity（2）就是基于这个库实现的。下面我们开始体验下如何使用这个库。

1，Replication

package main

import (
  "context"
  "os"
  "time"

  "github.com/go-mysql-org/go-mysql/mysql"
  "github.com/go-mysql-org/go-mysql/replication"
)

func main() {

  // Create a binlog syncer with a unique server id, the server id must be different from other MySQL's.
  // flavor is mysql or mariadb
  cfg := replication.BinlogSyncerConfig{
    ServerID: 100,
    Flavor:   "mysql",
    Host:     "127.0.0.1",
    Port:     3306,
    User:     "root",
    Password: "",
  }
  syncer := replication.NewBinlogSyncer(cfg)

  // Start sync with specified binlog file and position
  binlogFile := "/usr/local/var/mysql/mysql-bin.000020"
  binlogPos := uint32(1)
  streamer, _ := syncer.StartSync(mysql.Position{binlogFile, binlogPos})

  // or you can start a gtid replication like
  // streamer, _ := syncer.StartSyncGTID(gtidSet)
  // the mysql GTID set likes this "de278ad0-2106-11e4-9f8e-6edd0ca20947:1-2"
  // the mariadb GTID set likes this "0-1-100"

  for {
    ev, _ := streamer.GetEvent(context.Background())
    // Dump event
    ev.Dump(os.Stdout)
  }

  // or we can use a timeout context
  for {
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    ev, err := streamer.GetEvent(ctx)
    cancel()

    if err == context.DeadlineExceeded {
      // meet timeout
      continue
    }

    ev.Dump(os.Stdout)
  }
}

他的原理就是通过解析mysql的bin log来伪装成一个从库，实现主从复制。

2，Incremental dumping

package main

import (
  "github.com/go-mysql-org/go-mysql/canal"
  "github.com/siddontang/go-log/log"
)

type MyEventHandler struct {
  canal.DummyEventHandler
}

func (h *MyEventHandler) OnRow(e *canal.RowsEvent) error {
  log.Infof("%s %v\n", e.Action, e.Rows)
  return nil
}

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

func main() {
  cfg := canal.NewDefaultConfig()
  cfg.Addr = "127.0.0.1:3306"
  cfg.User = "root"
  // We only care table canal_test in test db
  cfg.Dump.TableDB = "test"
  cfg.Dump.Tables = []string{"canal_test"}

  c, err := canal.NewCanal(cfg)
  if err != nil {
    log.Fatal(err)
  }

  // Register a handler to handle RowsEvent
  c.SetEventHandler(&MyEventHandler{})

  // Start canal
  c.Run()
}

// create table canal_test (id int );
// Query OK, 0 rows affected (0.01 sec)

// insert into canal_test (id) values (12);
// Query OK, 1 row affected (0.00 sec)

// [2023/04/04 00:36:46] [info] binlogsyncer.go:813 rotate to (mysql-bin.000007, 4)
// [2023/04/04 00:36:46] [info] sync.go:63 received fake rotate event, next log name is mysql-bin.000007
// [2023/04/04 00:37:20] [info] main.go:13 insert [[12]]

它的原理和主从复制一样，只不过它是增量的复制。

3，Client

package main

import (
  "fmt"

  "github.com/go-mysql-org/go-mysql/client"
  "github.com/go-mysql-org/go-mysql/mysql"
)

func main() {

  // Connect MySQL at 127.0.0.1:3306, with user root, an empty password and database test
  conn, _ := client.Connect("127.0.0.1:3306", "root", "", "test")

  // Or to use SSL/TLS connection if MySQL server supports TLS
  //conn, _ := client.Connect("127.0.0.1:3306", "root", "", "test", func(c *Conn) {c.UseSSL(true)})

  // Or to set your own client-side certificates for identity verification for security
  //tlsConfig := NewClientTLSConfig(caPem, certPem, keyPem, false, "your-server-name")
  //conn, _ := client.Connect("127.0.0.1:3306", "root", "", "test", func(c *Conn) {c.SetTLSConfig(tlsConfig)})

  conn.Ping()

  // Insert
  r, _ := conn.Execute(`insert into table (id, name) values (1, "abc")`)

  // Get last insert id
  println(r.InsertId)
  // Or affected rows count
  println(r.AffectedRows)

  // Select
  var err error
  r, err = conn.Execute(`select id, name from table where id = 1`)

  fmt.Println(err)
  // Close result for reuse memory (it's not necessary but very useful)
  defer r.Close()

  // Handle resultset
  v, _ := r.GetInt(0, 0)
  fmt.Println(v)
  v, _ = r.GetIntByName(0, "id")

  // Direct access to fields
  for _, row := range r.Values {
    for _, val := range row {
      _ = val.Value() // interface{}
      // or
      if val.Type == mysql.FieldValueTypeFloat {
        _ = val.AsFloat64() // float64
      }
    }
  }

  // ...
  var result mysql.Result
  err = conn.ExecuteSelectStreaming(`select id, name from table LIMIT 100500`, &result, func(row []mysql.FieldValue) error {
    for idx, val := range row {
      field := result.Fields[idx]
      // You must not save FieldValue.AsString() value after this callback is done.
      // Copy it if you need.
      // ...
      fmt.Println(val, field)
    }
    return nil
  }, nil)

  // ...
}

既然实现了mysql协议，作为mysql客户端自然不在话下。

4，Fake server

package main

import (
  "log"
  "net"

  "github.com/go-mysql-org/go-mysql/server"
)

func main() {
  // Listen for connections on localhost port 4000
  l, err := net.Listen("tcp", "127.0.0.1:4000")
  if err != nil {
    log.Fatal(err)
  }

  // Accept a new connection once
  c, err := l.Accept()
  if err != nil {
    log.Fatal(err)
  }

  // Create a connection with user root and an empty password.
  // You can use your own handler to handle command here.
  conn, err := server.NewConn(c, "root", "", server.EmptyHandler{})
  if err != nil {
    log.Fatal(err)
  }

  // as long as the client keeps sending commands, keep handling them
  for {
    if err := conn.HandleCommand(); err != nil {
      log.Fatal(err)
    }
  }
}

它接受tcp请求，解析mysql协议，相应客户端请求，不过每个sql语句内部如何执行，需要自己定义。

4，Failover

package main

import (
  "context"

  "github.com/go-mysql-org/go-mysql/client"

  log "github.com/sirupsen/logrus"
)

func main() {
  pool := client.NewPool(log.Debugf, 100, 400, 5, "127.0.0.1:3306", `root`, ``, `test`)
  // ...
  ctx := context.Background()
  conn, _ := pool.GetConn(ctx)
  defer pool.PutConn(conn)

  conn.Execute("show databases;")
  conn.Begin()
  // etc...
}

通过选主的方式将从提升为主。

5，database/sql like driver

package main

import (
  "database/sql"

  _ "github.com/go-mysql-org/go-mysql/driver"
)

func main() {
  // dsn format: "user:password@addr?dbname"
  dsn := "root@127.0.0.1:3306?test"
  db, _ := sql.Open("mysql", dsn)
  db.Close()
}

完全兼容golang官方的mysql包提供的接口，可以作为mysql的驱动包使用。