Buriburizaemon/nezha_domains
1
0
Fork 0
mirror of https://github.com/Buriburizaem0n/nezha_domains.git synced 2026-02-06 13:40:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

theme-mdui by @MikoyChinese

Browse Source
This commit is contained in:
naiba
2022-01-09 11:54:14 +08:00
parent d1e81a2cde
commit 1e641d1249
21 changed files with 246 additions and 245 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
cmd/dashboard/rpc/rpc.go
View File

@@ -8,8 +8,8 @@ import (
"github.com/naiba/nezha/model"
pb "github.com/naiba/nezha/proto"
"github.com/naiba/nezha/service/dao"
rpcService "github.com/naiba/nezha/service/rpc"
"github.com/naiba/nezha/service/singleton"
)
func ServeRPC(port uint) {
@@ -29,45 +29,45 @@ func DispatchTask(serviceSentinelDispatchBus <-chan model.Monitor) {
for task := range serviceSentinelDispatchBus {
round := 0
endIndex := workedServerIndex
dao.SortedServerLock.RLock()
singleton.SortedServerLock.RLock()
// 如果已经轮了一整圈又轮到自己，没有合适机器去请求，跳出循环
for round < 1 || workedServerIndex < endIndex {
// 如果到了圈尾，再回到圈头，圈数加一，游标重置
if workedServerIndex >= len(dao.SortedServerList) {
if workedServerIndex >= len(singleton.SortedServerList) {
workedServerIndex = 0
round++
continue
}
// 如果服务器不在线，跳过这个服务器
if dao.SortedServerList[workedServerIndex].TaskStream == nil {
if singleton.SortedServerList[workedServerIndex].TaskStream == nil {
workedServerIndex++
continue
}
// 如果此任务不可使用此服务器请求，跳过这个服务器（有些 IPv6 only 开了 NAT64 的机器请求 IPv4 总会出问题）
if (task.Cover == model.MonitorCoverAll && task.SkipServers[dao.SortedServerList[workedServerIndex].ID]) ||
(task.Cover == model.MonitorCoverIgnoreAll && !task.SkipServers[dao.SortedServerList[workedServerIndex].ID]) {
if (task.Cover == model.MonitorCoverAll && task.SkipServers[singleton.SortedServerList[workedServerIndex].ID]) ||
(task.Cover == model.MonitorCoverIgnoreAll && !task.SkipServers[singleton.SortedServerList[workedServerIndex].ID]) {
workedServerIndex++
continue
}
// 找到合适机器执行任务，跳出循环
dao.SortedServerList[workedServerIndex].TaskStream.Send(task.PB())
singleton.SortedServerList[workedServerIndex].TaskStream.Send(task.PB())
workedServerIndex++
break
}
dao.SortedServerLock.RUnlock()
singleton.SortedServerLock.RUnlock()
}
}
func DispatchKeepalive() {
dao.Cron.AddFunc("@every 60s", func() {
dao.SortedServerLock.RLock()
defer dao.SortedServerLock.RUnlock()
for i := 0; i < len(dao.SortedServerList); i++ {
if dao.SortedServerList[i] == nil || dao.SortedServerList[i].TaskStream == nil {
singleton.Cron.AddFunc("@every 60s", func() {
singleton.SortedServerLock.RLock()
defer singleton.SortedServerLock.RUnlock()
for i := 0; i < len(singleton.SortedServerList); i++ {
if singleton.SortedServerList[i] == nil || singleton.SortedServerList[i].TaskStream == nil {
continue
}
dao.SortedServerList[i].TaskStream.Send(&pb.Task{Type: model.TaskTypeKeepalive})
singleton.SortedServerList[i].TaskStream.Send(&pb.Task{Type: model.TaskTypeKeepalive})
}
})
}