From a7e5430cf06a82f7c17b6565bd9f1ce581187adb Mon Sep 17 00:00:00 2001
From: zhangqian <zhangqian@123.com>
Date: 星期六, 21 十月 2023 15:24:58 +0800
Subject: [PATCH] 打印数据同步日志
---
serf/sync.go | 22 ++++++++++++++++++----
1 files changed, 18 insertions(+), 4 deletions(-)
diff --git a/serf/sync.go b/serf/sync.go
index ea21710..c92fe96 100644
--- a/serf/sync.go
+++ b/serf/sync.go
@@ -20,7 +20,7 @@
)
var (
- agent = SyncServer{}
+ agent = SyncServer{}
dependProcs = []string{
bhomeclient.Proc_System_Service,
}
@@ -124,7 +124,6 @@
}
}
if diff := arrayContains(existingProcs, dependProcs); diff == "" {
- initChan <- true
break loop
} else {
logx.Errorf("Proc: %s is not running!", diff)
@@ -372,7 +371,7 @@
}
func (ss *SyncServer) handleClusterMessage(msg []byte) {
- //fmt.Println("clusterMessage:", string(msg))
+ logx.Infof("clusterMessage:", string(msg))
sql := string(msg)
if len(sql) <= 0 {
@@ -403,8 +402,11 @@
}
}
+// 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 {
@@ -413,7 +415,19 @@
}
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
}
--
Gitblit v1.8.0