| | |
|---|
| | | } |
|---|
| | | } |
|---|
| | | |
|---|
| | | // serf 同步数据的限制为92160 byte |
|---|
| | | func (ss *SyncServer) handleSyncTableMessage(msg []byte) error { |
|---|
| | | sizeLimit := 61440 |
|---|
| | | targetId := string(msg) |
|---|
| | | |
|---|
| | | //fmt.Println("同步全量数据给节点:", targetId) |
|---|
| | | sqls, err := DumpTables(ss.sqlDB, ss.syncTables) |
|---|
| | | if err != nil { |
|---|
| | |
|---|
| | | } |
|---|
| | | |
|---|
| | | syncSql := strings.Join(sqls, ";") |
|---|
| | | err = ss.pubSyncSqlMessage([]byte(syncSql), targetId) |
|---|
| | | if len(syncSql) < sizeLimit { |
|---|
| | | err = ss.pubSyncSqlMessage([]byte(syncSql), targetId) |
|---|
| | | } else { |
|---|
| | | shard := "" |
|---|
| | | for _, sql := range sqls{ |
|---|
| | | if len(shard) + len(sql) > sizeLimit { |
|---|
| | | err = ss.pubSyncSqlMessage([]byte(shard), targetId) |
|---|
| | | shard = "" |
|---|
| | | } |
|---|
| | | |
|---|
| | | shard = fmt.Sprintf("%s%s;",shard,sql) |
|---|
| | | } |
|---|
| | | } |
|---|
| | | |
|---|
| | | return err |
|---|
| | | } |
|---|
