71 lines
1.8 KiB
Go
71 lines
1.8 KiB
Go
|
package trans
|
|||
|
|
|
|||
|
|
import (
|
|||
|
|
"context"
|
|||
|
|
"fmt"
|
|||
|
|
"os"
|
|||
|
|
"sort"
|
|||
|
|
"strings"
|
|||
|
|
"time"
|
|||
|
|
|
|||
|
|
"github.com/didi/nightingale/v5/config"
|
|||
|
|
"github.com/didi/nightingale/v5/models"
|
|||
|
|
"github.com/didi/nightingale/v5/naming"
|
|||
|
|
"github.com/didi/nightingale/v5/pkg/ipool"
|
|||
|
|
|
|||
|
|
"github.com/toolkits/pkg/logger"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
var connPools *ipool.ConnPools
|
|||
|
|
var svcsCache string
|
|||
|
|
|
|||
|
|
func Start(ctx context.Context) {
|
|||
|
|
// 初始化本包的数据结构，然后启动一个goroutine，周期性获取活着的judge实例，更新相应的pool、queue等
|
|||
|
|
judgeConf := config.Config.Judge
|
|||
|
|
connPools = ipool.NewConnPools(judgeConf.ConnMax, judgeConf.ConnIdle, judgeConf.ConnTimeout, judgeConf.CallTimeout, []string{})
|
|||
|
|
|
|||
|
|
if err := syncInstances(); err != nil {
|
|||
|
|
fmt.Println("syncInstances fail:", err)
|
|||
|
|
logger.Close()
|
|||
|
|
os.Exit(1)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
go loopSyncInstances()
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func loopSyncInstances() {
|
|||
|
|
interval := time.Duration(config.Config.Heartbeat.Interval) * time.Millisecond
|
|||
|
|
for {
|
|||
|
|
time.Sleep(interval)
|
|||
|
|
if err := syncInstances(); err != nil {
|
|||
|
|
logger.Warning("syncInstances fail:", err)
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func syncInstances() error {
|
|||
|
|
// 获取当前活着的所有实例
|
|||
|
|
instances, err := models.InstanceGetAlive(config.EndpointName)
|
|||
|
|
if err != nil {
|
|||
|
|
logger.Warningf("mysql.error: get alive server instances fail: %v", err)
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// 排序，便于与内存中的实例列表做差别判断
|
|||
|
|
sort.Strings(instances)
|
|||
|
|
|
|||
|
|
// 如果列表变化，就去处理，并且要更新内存变量serverStr
|
|||
|
|
newSvcs := strings.Join(instances, ",")
|
|||
|
|
if newSvcs != svcsCache {
|
|||
|
|
// 如果有新实例，创建对应的连接池，如果实例少了，删掉没用的连接池
|
|||
|
|
connPools.UpdatePools(instances)
|
|||
|
|
// 如果有新实例，创建对应的Queue，如果实例少了，删掉对应的Queue
|
|||
|
|
queues.Update(instances)
|
|||
|
|
// 重建哈希环
|
|||
|
|
naming.RebuildConsistentHashRing(instances)
|
|||
|
|
svcsCache = newSvcs
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
return nil
|
|||
|
|
}
|