写数据库同步 binlog #5

wenbobuaa · 2018-10-30T12:34:19Z

从一机器读取 mysql binlog 数据向另一数据库执行 sql 语句进行 binlog 同步。

使用 go 实现，依赖的 go-mysql 包比较大，没有在这个目录。

详细描述： #2

drmingdrmer · 2018-10-30T13:00:10Z

go/src/binlogsync/binlog_sync.go

+
+type DBInfo struct {
+	Conn  *client.Conn
+	Shard *DBShard


1个connection只有1个shard?

这里是一条 sql 记录对应一个 DBInfo，包括它所属的 shard 和这个 shard 对应的connection

drmingdrmer · 2018-10-30T13:00:37Z

go/src/binlogsync/binlog_sync.go

+
+type OutStatus struct {
+	err          error
+	routineIndex int


routine中文是什么意思?

goroutine 类似线程

叫goroutine吧...routine的解释有点多

drmingdrmer · 2018-10-30T13:01:20Z

go/src/binlogsync/binlog_sync.go

+type OutStatus struct {
+	err          error
+	routineIndex int
+	logPos       uint32


这里如果可以的话用有符号整数. 所有代码中避免使用无符号数, 一遇到减法就容易溢出

drmingdrmer · 2018-10-30T13:01:59Z

go/src/binlogsync/binlog_sync.go

+
+type Config struct {
+	ChCap int
+	ChCnt int


这2个变量不造是啥意思...配置项当做全局变量,尽量用完整单词.

好是指的 channel 的数量和容量。这里改一下

所有的channel不放在一个slice里吗?如果放的话用slice的len维护数量就行了...

这里想的在配置文件里指定一下需要创建多少个 channel

drmingdrmer · 2018-10-30T13:03:30Z

go/src/binlogsync/binlog_sync.go

+	TableIndex []string
+
+	BinlogFile string
+	BinlogPos  uint32


咱们应该都是用的gtid来追踪binlog的. 文件+pos的方式在多个源上不一样.不太好处理.
不造你这里为啥是要加这2个东西

之前参考的python版的同步用的是文件+pos 的方式，这里也用这样来读binlog 的。我再看看 gtid 的方式

恩...好像读的话还必须对源还必须制定binlog的file+pos, 不造有没有通过gtid来指定的方式...

有的增加了一个根据 gtid 读

drmingdrmer

writeThreadCount 跟 ChannelCapacity 是啥关系...在这里跟整个代码的逻辑的关系是啥...加点注释解释下吧...

drmingdrmer · 2018-10-31T06:57:41Z

go/src/binlogsync/binlog_sync.go

-	ChCap int
-	ChCnt int
+	ChannelCapacity  int
+	WriteThreadCount int


咱们代码里好像count都用的cnt...

好的我加点注释

thread 似乎不准确这里用的是goroutine 吧？这里用来控制并发吧，而且控制并发数就这一个变量，干脆叫 worker 怎么样？

这里的 Config 是用来解 binary -> json 的吧。最好定义下一下 json 字段名

drmingdrmer · 2018-10-31T07:05:02Z

go/src/binlogsync/binlog_sync.go

+	FileLog = log.New(logFile, "", log.Ldate|log.Ltime|log.Lshortfile)
+
+	// read config
+	jsonParser := NewJsonStruct()


配置用yaml, rpc消息用json. 分别偏重可读性和解析的效率.

配置允许用json写, 但用yaml解析. 这样支持可读性更好的yaml配置.

好的这里修改一下

drmingdrmer · 2018-10-31T07:11:57Z

go/src/binlogsync/util.go

+
+func calcHashToInt64(src []byte) (int64, error) {
+	h := sha256.New()
+	h.Write(src)


计算hash时, 因为不是为了校验, 只是为了sharding, 可以用弱1点的, 只要差不多均匀就可以了. @templexxx 那应该调研过很多快速的hash. 求1个够快够均匀的, 输出是整数就行.

这个环节可以考虑降低点cpu开销, 字符串不要拼接, 一个一个update到hash函数里, 类似sha1:init(); sha1:update(buf); sha1:final() 酱.

好的这里重新写一下

看上去 murmurhash fnv adler32 包括 crc32 应该都 okay

drmingdrmer · 2018-10-31T07:12:37Z

go/src/binlogsync/sqlUtil.go

+}
+
+func quote(src, quote string) string {
+	rst := strings.Replace(src, quote, "\\"+quote, -1)


参数跟函数名一样不太好...

drmingdrmer · 2018-10-31T07:15:54Z

go/src/binlogsync/sqlUtil.go

+}
+
+func quote(src, quote string) string {
+	rst := strings.Replace(src, quote, "\\"+quote, -1)


quote必须也replace \, 否则mysql解析出错. 这类基础操作涉及到很多细节, 需要测试. 另外没有现成的 blabla-escape的函数用咩?

` --> \` \` --> \\\` ## 不应该是 \\`

有 escape 函数。这里 quote 的作用只是用 " 把 src 括起来。好像也不太需要这个函数，escape 之后直接 + " 就好...

what? 不是用来替换table名字中的反引号的吗?

嗯对这个函数还是有用我修改一下

sejust · 2018-10-31T07:32:58Z

go/src/binlogsync/config.json

+    "TableField": ["bucket_id", "scope", "key", "ts", "is_del", "owner", "acl", "sha1",
+        "ver", "md5", "crc32", "size", "file_meta", "group_id", "origo", "expires", "multipart"],
+    "TableShard": [ "bucket_id" , "scope", "key"],
+    "TableIndex": [ "bucket_id" , "scope", "key"],


index 应该是 bucket_id, scope, key, ts

sejust · 2018-10-31T07:53:21Z

go/src/binlogsync/binlog_sync.go

+
+func writeToDB(chIdx int, inCh chan *WriteEvent, outCh chan *OutStatus) {
+	for ev := range inCh {
+		mutex.Lock()


这里需要lock吗？

getEventConnection 里面查找 shard 时有一段二分查找的逻辑，会有值的修改和比较需要锁上

二分查找为啥会有值的修改?

是根据下标找 slice 里的 shard ，这个 shard 是全局的，过程中下标的修改可能导致找到错误的 shard 吧

这里是不用加锁

sejust · 2018-10-31T08:13:41Z

go/src/binlogsync/binlog_sync.go

+			writeEV := &WriteEvent{
+				event: ev,
+			}
+			writeChs[0] <- writeEV


writeToDB 这个函数应该处理不了 replication.ROTATE_EVENT 这个event

对的。处理 rotate 和记录已经同步到的位置和恢复执行要再考虑下，这部分后面更新一个提交点。

templexxx · 2018-10-31T08:19:13Z

参考一下 ec 项目 go 的文件布局调整一下吧。在这里 binlogsync 就是一个单纯的 app，不需要外部引用他吧。那么建立一个 app 目录，下面再建立一个 binlogsync 目录用来存放你的文件。下面包含： binlogsync.go binlogsync.conf（作为配置示例） sql.go util.go

templexxx · 2018-10-31T08:20:37Z

go/src/binlogsync/binlog_sync.go

+func main() {
+
+	// set log
+	ShellLog = log.New(os.Stdout, "", 0)


ShellLog 为啥大写？

好像没啥特殊的，就用小写吧

templexxx · 2018-10-31T08:21:14Z

go/src/binlogsync/binlog_sync.go

+	// set log
+	ShellLog = log.New(os.Stdout, "", 0)
+
+	logFile, err := os.Create(logFileName)


这里覆盖了你定义的全局 logFile

是.. 改一下..

templexxx · 2018-10-31T08:32:13Z

go/src/binlogsync/binlog_sync.go

+var dbPool = make(map[string]*client.Conn)
+var conf = Config{}
+var logFileName = "binlog_sync.out"
+var confName = "./config.json"


logFileName confName 还是作为命令行参数传进来吧，方便调试

嗷嗷可以改一下

templexxx · 2018-10-31T08:32:54Z

go/src/binlogsync/binlog_sync.go

+)
+
+var dbPool = make(map[string]*client.Conn)
+var conf = Config{}


var conf Config 应该就可以了

templexxx · 2018-10-31T08:34:16Z

go/src/binlogsync/util.go

+type JsonStruct struct {
+}
+
+func NewJsonStruct() *JsonStruct {


为啥要定义这么一个数据结构？没意义啊？

看起来是多余这里要重写下

templexxx · 2018-10-31T08:36:12Z

go/src/binlogsync/binlog_sync.go

-	ChCap int
-	ChCnt int
+	ChannelCapacity  int
+	WriteThreadCount int


thread 似乎不准确这里用的是goroutine 吧？这里用来控制并发吧，而且控制并发数就这一个变量，干脆叫 worker 怎么样？

templexxx · 2018-10-31T08:37:19Z

go/src/binlogsync/binlog_sync.go

-	ChCap int
-	ChCnt int
+	ChannelCapacity  int
+	WriteThreadCount int


这里的 Config 是用来解 binary -> json 的吧。最好定义下一下 json 字段名

templexxx · 2018-10-31T09:07:59Z

go/src/binlogsync/binlog_sync.go

+}
+
+type Config struct {
+	ChannelCapacity  int


为啥 channelCapacity 要做为参数？

templexxx · 2018-10-31T09:11:50Z

go/src/binlogsync/binlog_sync.go

+	return &conf.Shards[0]
+}
+
+func main() {


用 waitGroup 保证一下执行完成

记得 close chanel

主线程收集处理结果，其他交由 goroutine 处理。

templexxx · 2018-10-31T09:28:31Z

go/src/binlogsync/binlog_sync.go

+
+var logFile *os.File
+
+var mutex sync.Mutex


锁不用引用类型感觉怪怪得

噢用引用吧

drmingdrmer · 2018-10-31T13:24:40Z

go/src/binlogsync/sqlUtil.go

-	rst := strings.Replace(src, quote, "\\"+quote, -1)
+func quoteString(src, quote string) string {
+	safeQuote := strings.Replace(quote, "\\", "\\\\", -1)
+	rst := strings.Replace(src, quote, "\\"+safeQuote, -1)


感觉逻辑不太对...替换quote里的\是啥意思

嗯.. 感觉这里应该限制 len(quote) == 1 ，这样只用替换 src 中的quote就好了。len(quote) > 1 的话不太好处理replace src 里的 quote

替换quote里的\是啥意思

为了解决这种情况出现, 但是没想对..

` -> \` \` -> \\\` # 不是 \\`

drmingdrmer · 2018-10-31T13:26:04Z

go/src/binlogsync/binlog_sync.go

+	i := sort.Search(len(conf.Shards), func(i int) bool {
 		shard := conf.Shards[i].From
-
 		rst, err := compareStringSlice(shard, tbShards)


额...row里的所有field都是string? 没有整数吗?

从binlog 读出来后，都统一变成了 string 处理

templexxx · 2018-11-01T06:32:13Z

go/src/binlogsync/binlog_sync.go

+func main() {
+
+	// set log
+	shellLog = log.New(os.Stdout, "", 0)


似乎这里的 shellLog 的用处可以直接用 print 代替

还是加个shelllog吧.以后替换容易点...最终肯定不会往shell里打印吧...

drmingdrmer · 2018-11-01T07:21:20Z

go/src/binlogsync/binlog_sync.go

+var (
+	mutex *sync.Mutex
+
+	wg sync.WaitGroup


以面向对象的方式来实现吧, 一个机器上肯定不止1个复制关系在跑, 可能一个进程里要处理多个复制.

没有明白过去问问你

drmingdrmer · 2018-11-01T07:22:40Z

go/src/binlogsync/binlog_sync.go

+)
+
+func validRow(srcRow []interface{}) map[string]string {
+	// the first column `id` should not put in new rowValue


不太明白, 都处理成string的好处是啥?

函数名不太合适, normalize 好点...valid是形容词

都变成 string 方便一个row放在一个map。保留原格式像 int []byte 好像也没特别用处..

wenbobuaa added 3 commits October 30, 2018 20:21

make sql statement

93b2c11

add some utils

9d7f2f5

add binlog sync

dcc5d59

wenbobuaa self-assigned this Oct 30, 2018

wenbobuaa requested review from drmingdrmer, sejust and templexxx October 30, 2018 12:34

drmingdrmer reviewed Oct 30, 2018

View reviewed changes

add read binlog by GTID

84db494

drmingdrmer reviewed Oct 31, 2018

View reviewed changes

add some commends

5552286

sejust reviewed Oct 31, 2018

View reviewed changes

templexxx reviewed Oct 31, 2018

View reviewed changes

refine after review

1fc8d2e

drmingdrmer reviewed Oct 31, 2018

View reviewed changes

templexxx reviewed Nov 1, 2018

View reviewed changes

drmingdrmer reviewed Nov 1, 2018

View reviewed changes

wenbobuaa mentioned this pull request Nov 8, 2018

以写数据库的方式同步 binlog #7

Open

写数据库同步 binlog #5

Are you sure you want to change the base?

写数据库同步 binlog #5

Uh oh!

Conversation

wenbobuaa commented Oct 30, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

drmingdrmer left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!