update

zhangqian

2023-10-27 d91a802e7aa2ad4075ed803b8ddc7536a91a0ef0

 serf/sync.go

@@ -371,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 {
@@ -410,22 +410,27 @@
   //fmt.Println("同步全量数据给节点:", targetId)
   sqls, err := DumpTables(ss.sqlDB, ss.syncTables)
   if err != nil {
      fmt.Println("DumpTables error, ", err.Error())
      logx.Errorf("DumpTables error: %s", err.Error())
      return err
   }

   logx.Infof("DumpTables sql:%v", sqls)
   syncSql := strings.Join(sqls, ";")
   if len(syncSql) < sizeLimit {
      err = ss.pubSyncSqlMessage([]byte(syncSql), targetId)
   } else {
      shard := ""
      for _, sql := range sqls{
         if len(shard) + len(sql) > sizeLimit {
      for _, sql := range sqls {
         if len(shard)+len(sql) > sizeLimit {
            err = ss.pubSyncSqlMessage([]byte(shard), targetId)
            shard = ""
         }

         shard = fmt.Sprintf("%s%s;",shard,sql)
         shard = fmt.Sprintf("%s%s;", shard, sql)
      }

      if len(shard) >0 {
         err = ss.pubSyncSqlMessage([]byte(shard), targetId)
      }
   }