forked from sunface/rust-by-practice
-
Notifications
You must be signed in to change notification settings - Fork 0
/
request.go
160 lines (146 loc) · 4 KB
/
request.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
package dotray
import (
"encoding/gob"
"fmt"
"net"
"strconv"
)
// Request 节点之间交换的数据结构
type Request struct {
ID int64
Command int
Data interface{}
From string
}
const (
NormalRequest = 0 // 正常请求,发送数据报文
NormalRequestReceived = 1 // 消息已经接收ack
ServersRequest = 2 // 请求服务器列表
ServerResponse = 3 // 返回服务器列表
ServerPing = 4 // 发送节点的监听地址
ServerPong = 5 // ping请求的ack
BackupSeeds = 6 // 备用种子节点列表
SyncBackupSeeds = 7 // 请求获取备用种子列表
)
func (r *Request) handle(node *Node, conn net.Conn) (string, error) {
switch r.Command {
case NormalRequestReceived:
// 从待重发队列中,删除消息
deleteResend(r.ID, r.From)
case NormalRequest:
// 接收到远程节点发来的消息
//发送给本地节点处理
//转发给种子节点和下游节点
routeSend(node, r)
// 回复消息收到的ack
encoder := gob.NewEncoder(conn)
encoder.Encode(Request{
ID: r.ID,
Command: NormalRequestReceived,
From: node.nodeAddr,
})
case SyncBackupSeeds:
// 请求节点的地址
fromAddr := r.Data.(string)
// 请求获取备用种子列表
//从当前种子 + 下游列表中选出合适的种子节点(根据负载)
// 要避免互为种子的情况
var addrs []string
if node.seedAddr != "" && node.seedAddr != fromAddr {
addrs = append(addrs, node.seedAddr)
}
for addr := range node.downstreams {
if len(addrs) < maxBackupSeedLen && addr != fromAddr {
addrs = append(addrs, addr)
}
}
encoder := gob.NewEncoder(conn)
encoder.Encode(Request{
Command: BackupSeeds,
Data: addrs,
})
case BackupSeeds:
addrs := r.Data.([]string)
// 当前种子节点发送备用种子节点
for _, addr1 := range addrs {
if addr1 == "" {
continue
}
// 种子更新策略
//1.备用节点数未达上限,添加到列表
//2.达到上限,替换重试次数超过seedMaxRetry,且从最高的开始替换
exist := false
maxRetry := 0
for _, seed := range node.seedBackup {
if seed.retry > maxRetry {
maxRetry = seed.retry
}
if addr1 == seed.addr {
exist = true
break
}
}
// 新的种子在当前的备用列表中不存在
if !exist {
if len(node.seedBackup) >= maxBackupSeedLen {
// 若备用列表中的种子最大重试次数没有超过阀值,则不替换
if maxRetry <= seedMaxRetry {
break
}
// 替换第一个超过阀值的旧种子
for i, seed := range node.seedBackup {
if seed.retry > seedMaxRetry {
node.seedBackup[i] = &Seed{
addr: addr1,
retry: 0,
}
}
}
} else {
// 添加新种子
node.seedBackup = append(node.seedBackup, &Seed{
addr: addr1,
retry: 0,
})
}
}
}
// 打印当前种子、备份种子、下游节点
// fmt.Printf("当前种子:%s,备份种子:%v,下游节点:%v\n", node.seedAddr, getSeedAddrs(node.seedBackup), node.downstreams)
case ServerPing:
// 下游节点发送它的监听地址
addr, ok := r.Data.(string)
if ok {
// 添加进本节点下游节点列表
lock.Lock()
node.downstreams[addr] = conn
lock.Unlock()
// 添加进下游节点后,目标节点不得在种子备份列表中存在
for i, seed := range node.seedBackup {
if seed.addr == addr {
node.seedBackup = append(node.seedBackup[:i], node.seedBackup[i+1:]...)
break
}
}
// 返回Pong
encoder := gob.NewEncoder(conn)
encoder.Encode(Request{
Command: ServerPong,
From: node.nodeAddr,
})
return addr, nil
}
case ServerPong:
node.pinged = true
default:
fmt.Println("未识别的消息类型:", r.Command)
}
return "", nil
}
func getSeedAddrs(seeds []*Seed) []string {
addrs := make([]string, len(seeds))
for i, seed := range seeds {
addrs[i] = seed.addr + "/" + strconv.Itoa(seed.retry)
}
return addrs
}