以写数据库的方式同步 binlog #7
Conversation
go/src/app/binlogsync/binlog_sync.go
Outdated
| var errTypes = make(map[string]int) | ||
|
|
||
| var start = time.Now() | ||
| for outStat := range bs.CountCh { |
There was a problem hiding this comment.
这里的统计没有保证 先读的 event 先被收集到，记录的binlog位置会有问题，需要修改..
go/src/app/binlogsync/binlog_sync.go
Outdated
| event *replication.BinlogEvent | ||
| before map[string]interface{} | ||
| after map[string]interface{} | ||
| dbInfo *DBInfo |
go/src/app/binlogsync/binlog_sync.go
Outdated
| type OutMessage struct { | ||
| Synced int64 `yaml:"Synced"` | ||
| Faild int64 `yaml:"Faild"` | ||
| Rate float64 `yaml:"Rate(rows/s)"` |
go/src/app/binlogsync/util.go
Outdated
| lenA := len(a) | ||
| lenB := len(b) | ||
| if lenA != lenB { | ||
| return 0, errors.New(fmt.Sprintf("length of slices not equals: %d != %d", lenA, lenB)) |
There was a problem hiding this comment.
脱离这个场景, 类似python里, slice长度不同是可以比较的
go/src/app/binlogsync/util.go
Outdated
| var rst int | ||
| switch v.(type) { | ||
| case int: | ||
| ai := reflect.ValueOf(v).Int() |
There was a problem hiding this comment.
这里为啥不直接v.(int) reflect好像挺慢啊...
There was a problem hiding this comment.
这里 v.(type) case 到的 int 包括了 int32 int64 等，但是 v.(int) 只能转化 v 是 int 类型的，不能转化 v 是 int64 这种类型的.. 所以才想用 reflect 都会变成 int64
There was a problem hiding this comment.
那...为啥不分布case int 和int64...
There was a problem hiding this comment.
这样 case 太多，还有一点是这样分 case 的话 a[i].(int) ，b[i].(int64) 不在一个 case 还要其他步骤找出来比，已知问题 yaml 把interface 类型的数字读成 int 类型，binlog 里的数字是 int64... 所以用了reflect 看起来整齐点...
There was a problem hiding this comment.
reflect可能会影响性能...
case可以同时捕捉int和int64吗? 都转int64咋样...
go/src/app/binlogsync/util.go
Outdated
| } | ||
| } | ||
|
|
||
| func parseYAML(filename string, v interface{}) error { |
There was a problem hiding this comment.
要不就符合go风格, marshal unmarshal, 或者顺着py风格, load dump
go/src/app/binlogsync/binlog_sync.go
Outdated
| if rst <= 0 { | ||
| return true | ||
| } | ||
| return false |
There was a problem hiding this comment.
直接return rst <= 0 好不...
go/src/app/binlogsync/binlog_sync.go
Outdated
| } | ||
|
|
||
| func (bs *BinlogSyncer) formatRow(srcRow []interface{}) map[string]interface{} { | ||
| // the first column `id` should not put in new rowValue |
There was a problem hiding this comment.
为啥should not...解释下...我觉得应该放进去啊...
There was a problem hiding this comment.
想的是 这个 id 是自增的一列 ，写语句里不加这行让目标表自增这列
There was a problem hiding this comment.
binlog复制应该完整复制数据. id只有在咱们的场景里是不用的. 其他地方可能是有逻辑依赖的.
go/src/app/binlogsync/binlog_sync.go
Outdated
|
|
||
| bs.mutex.Lock() | ||
| _, err = ev.dbInfo.Conn.Execute(sql) | ||
| bs.mutex.Unlock() |
There was a problem hiding this comment.
这里的Conn 是几个线程公用的，Execute 实现里面有一个 setSequence 的操作会修改 Conn 的值, 测试出错了看到的
| whereClause = makeWhereClause(idxField, idxValue) | ||
| limitClause = "LIMIT 1" | ||
|
|
||
| return fmt.Sprintf("UPDATE %s SET %s%s%s;", tableClause, setClause, whereClause, limitClause) |
There was a problem hiding this comment.
恩...update 如果update的是index fields里的内容这里能处理吗...
There was a problem hiding this comment.
但是.源库的update不一定是根据index来update的吧...可能只根据primary key的id来update...
update语句在apply binlog时应该是必须用1个delete加1个insert才能完整复现update的逻辑.
况且, 考虑binlog重放, 第2次update这个东西可能会失败.
wenbobuaa
force-pushed
the
binlog-sync
branch
from
2cd6ba3
to
97cf753
Compare
wenbobuaa
force-pushed
the
binlog-sync
branch
from
97cf753
to
1f06031
Compare
和 #5 是一个事儿，那个评论有点多。改动挺大，新开了PR方便看。
从一机器读取 mysql binlog 数据向另一数据库执行 sql 语句进行 binlog 同步。
使用 go 实现，依赖 的 go-mysql 包比较大，目前没有在这个目录。
详细描述： #2