fix

zhangqian

2023-10-30 10c65da3d2af7056f48d9301e83f53f102f76e18

 serf/sync.go

@@ -1,6 +1,7 @@
package serf

import (
   "apsClient/pkg/logx"
   "context"
   "encoding/json"
   "fmt"
@@ -19,7 +20,10 @@
)

var (
   agent = SyncServer{}
   agent       = SyncServer{}
   dependProcs = []string{
      bhomeclient.Proc_System_Service,
   }
)

const (
@@ -101,6 +105,42 @@
   bhomedbapi.InitDoReq(client.RequestOnly)
   //bhomedbapi.InitLog(logger.Debug)

   // 需要等待system-service进程成功启动后，才能获取集群状态(或者保证程序启动时获取到正确的状态)
   tryTimes := 0
loop:
   for {
      select {
      case <-q:
         initChan <- false
         return
      default:
         if tryTimes < 15 {
            clients, err := client.GetRegisteredClient()
            if err == nil && len(clients) > 0 {
               var existingProcs []string
               for _, c := range clients {
                  if c.Online {
                     existingProcs = append(existingProcs, string(c.Proc.ProcId))
                  }
               }
               if diff := arrayContains(existingProcs, dependProcs); diff == "" {
                  break loop
               } else {
                  logx.Errorf("Proc: %s is not running!", diff)
                  time.Sleep(time.Second * 1)
               }
            } else {
               tryTimes++
               time.Sleep(time.Second * 5)
            }
         } else {
            logx.Errorf("tried 15 times, client.GetRegisteredClient failed")
            initChan <- false
            return
         }
      }
   }

   go client.StartServer(nil)

   ss.bhClient = client
@@ -112,9 +152,9 @@
   // 启动后查询一次集群状态
   ss.QueryClusterStat()

   if ss.ClusterStatus != "" {
      ss.sqlDB.LogMode(true)
   }
   //if ss.ClusterStatus != "" {
   ss.sqlDB.LogMode(true)
   //}

   initChan <- true
   <-q
@@ -157,7 +197,7 @@
      return err
   }

   fmt.Println("加入集群, 请求同步全量数据,id:", ss.ServerId)
   logx.Debugf("加入集群, 请求同步全量数据,id:%s", ss.ServerId)
   return ss.bhClient.Publish(serfSyncTopic, b)
}

@@ -177,7 +217,7 @@
         // 处理同步全量数据的请求
         if string(busMsg.Topic) == ss.queryTableTopic {
            if ss.ClusterStatus == "master" {
               fmt.Println("接收到同步全量数据请求")
               logx.Debugf("接收到同步全量数据请求.")
               ss.handleSyncTableMessage(busMsg.Data)
            }
         }
@@ -212,7 +252,15 @@
                  // 退出集群, 开启日志跟踪, 设置角色slave
                  ss.clusterEventFn(EventLeaveCluster)
                  ss.ClusterStatus = ""
                  ss.sqlDB.LogMode(false)
                  ss.sqlDB.LogMode(true)
               case "slave2master":
                  ss.clusterEventFn(EventSlave2Master)
                  ss.ClusterStatus = "master"
                  ss.sqlDB.LogMode(true)
               case "master2slave":
                  ss.clusterEventFn(EventMaster2Slave)
                  ss.ClusterStatus = "slave"
                  ss.sqlDB.LogMode(true)
               }
            }
         }
@@ -240,7 +288,7 @@

      err = tx.Exec(delSql).Error
      if err != nil {
         fmt.Println("删除本地的同步库数据失败,", err.Error())
         logx.Errorf("删除本地的同步库数据失败, %s", err.Error())
      }
   }

@@ -277,7 +325,7 @@

   ss.ClusterStatus = reply.Msg

   fmt.Println("当前集群状态:", ss.ClusterStatus)
   logx.Debugf("当前集群状态: %s", ss.ClusterStatus)

   return reply.Msg
}
@@ -323,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 {
@@ -354,17 +402,59 @@
   }
}

// serf 同步数据的限制为92160 byte
func (ss *SyncServer) handleSyncTableMessage(msg []byte) error {
   sizeLimit := 61440
   targetId := string(msg)
   fmt.Println("同步全量数据给节点:", targetId)

   //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, ";")
   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)
      }

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

   return err
}

func arrayContains(list []string, arr []string) string {
   if arr == nil || list == nil {
      return ""
   }

   for _, s := range arr {
      isExist := false
      for _, t := range list {
         if s == t {
            isExist = true
            break
         }
      }

      if !isExist {
         return s
      }
   }

   return ""
}