-
Notifications
You must be signed in to change notification settings - Fork 97
/
rpc.invoker.go
247 lines (224 loc) · 6.11 KB
/
rpc.invoker.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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
package rpc
import (
"fmt"
"strings"
"time"
"github.com/micro-plat/hydra/rpc/balancer"
"github.com/micro-plat/lib4go/concurrent/cmap"
"github.com/micro-plat/lib4go/logger"
)
//Invoker RPC服务调用器,封装基于域及负载算法的RPC客户端
type Invoker struct {
cache cmap.ConcurrentMap
address string
opts []ClientOption
domain string
server string
lb balancer.CustomerBalancer
*invokerOption
}
type invokerOption struct {
logger *logger.Logger
timerout time.Duration
// balancerType int
balancers map[string]BalancerMode
servers string
// localPrefix string
tls map[string][]string
}
type BalancerMode struct {
Mode int
Param string
}
const (
//RoundRobin 轮询负载算法
RoundRobin = iota + 1
//LocalFirst 本地优先负载算法
LocalFirst
)
//InvokerOption 客户端配置选项
type InvokerOption func(*invokerOption)
//WithInvokerLogger 设置日志记录器
func WithInvokerLogger(log *logger.Logger) InvokerOption {
return func(o *invokerOption) {
o.logger = log
}
}
func WithBalancerMode(platName string, mode int, p string) InvokerOption {
switch mode {
case RoundRobin:
return WithRoundRobin(platName)
case LocalFirst:
return WithLocalFirst(p, platName)
default:
return func(o *invokerOption) {
}
}
}
//WithRoundRobin 设置为轮询负载
func WithRoundRobin(platName ...string) InvokerOption {
return func(o *invokerOption) {
if len(platName) == 0 {
o.balancers["*"] = BalancerMode{Mode: RoundRobin}
return
}
for _, v := range platName {
o.balancers[v] = BalancerMode{Mode: RoundRobin}
}
}
}
//WithLocalFirst 设置为本地优先负载
func WithLocalFirst(prefix string, platName ...string) InvokerOption {
return func(o *invokerOption) {
if prefix != "" {
if len(platName) == 0 {
o.balancers["*"] = BalancerMode{Mode: LocalFirst, Param: prefix}
return
}
for _, v := range platName {
o.balancers[v] = BalancerMode{Mode: LocalFirst, Param: prefix}
}
}
}
}
//WithRPCTLS 设置TLS证书(pem,key)
func WithRPCTLS(platName string, tls []string) InvokerOption {
return func(o *invokerOption) {
if len(tls) == 2 {
o.tls[platName] = tls
}
}
}
//NewInvoker 构建RPC服务调用器
//domain: 当前服务所在域
//server: 当前服务器名称
//addrss: 注册中心地址格式: zk://192.168.0.1166:2181或standalone://localhost
func NewInvoker(domain string, server string, address string, opts ...InvokerOption) (f *Invoker) {
f = &Invoker{
domain: domain,
server: server,
address: address,
cache: cmap.New(8),
invokerOption: &invokerOption{
balancers: map[string]BalancerMode{
"*": BalancerMode{Mode: RoundRobin},
},
tls: make(map[string][]string),
},
}
for _, opt := range opts {
opt(f.invokerOption)
}
if f.invokerOption.logger == nil {
f.invokerOption.logger = logger.GetSession("rpc.invoker", logger.CreateSession())
}
return
}
//RequestFailRetry 失败重试请求
func (r *Invoker) RequestFailRetry(service string, method string, header map[string]string, form map[string]interface{}, times int) (status int, result string, params map[string]string, err error) {
for i := 0; i < times; i++ {
status, result, params, err = r.Request(service, method, header, form, true)
if err == nil || status < 500 {
return
}
}
return
}
//Request 使用RPC调用Request函数
func (r *Invoker) Request(service string, method string, header map[string]string, form map[string]interface{}, failFast bool) (status int, result string, params map[string]string, err error) {
status = 500
client, err := r.GetClient(service)
if err != nil {
return
}
_, rservice, d, s, _ := ResolvePath(service, r.domain, r.server)
status, result, params, err = client.Request(rservice, method, header, form, failFast)
if status != 200 || err != nil {
if err != nil {
err = fmt.Errorf("%s[@%s.%s]请求失败:%v(%d)", rservice, d, s, err, status)
} else {
err = fmt.Errorf("%s[@%s.%s]请求失败:%d)", rservice, d, s, status)
}
}
return
}
func (r *Invoker) getBalancer(domain string) (int, string) {
if b, ok := r.balancers[domain]; ok {
return b.Mode, b.Param
}
if b, ok := r.balancers["*"]; ok {
return b.Mode, b.Param
}
return RoundRobin, ""
}
//GetClient 获取RPC客户端
//addr 支持格式:
//order.request@merchant.hydra
//order.request,order.request@api.hydra
//order.request@api
func (r *Invoker) GetClient(addr string) (c *Client, err error) {
isIP, rservice, domain, server, err := ResolvePath(addr, r.domain, r.server)
if err != nil {
return
}
//
serviceKey := fmt.Sprintf("%s@%s.%s", rservice, server, domain)
_, client, err := r.cache.SetIfAbsentCb(serviceKey, func(i ...interface{}) (interface{}, error) {
plat := i[0].(string)
server := i[1].(string)
service := i[2].(string)
isip := i[3].(bool)
opts := make([]ClientOption, 0, 0)
opts = append(opts, WithLogger(r.logger))
//IP直接调用
if isip {
//设置安全证书
switch len(r.tls[domain]) {
case 2:
opts = append(opts, WithTLS(r.tls[domain]))
}
return NewClient(server, opts...)
}
//非IP调用
mode, p := r.getBalancer(domain)
rs := balancer.NewResolver(plat, server, service, time.Second, p)
servicePath := fmt.Sprintf(serviceRoot, strings.TrimPrefix(domain, "/"), server, service)
//设置负载均衡算法
switch mode {
case RoundRobin:
opts = append(opts, WithRoundRobinBalancer(rs, servicePath, time.Second, map[string]int{}))
case LocalFirst:
opts = append(opts, WithLocalFirstBalancer(rs, servicePath, p, map[string]int{}))
default:
}
//设置安全证书
switch len(r.tls[domain]) {
case 2:
opts = append(opts, WithTLS(r.tls[domain]))
}
return NewClient(r.address, opts...)
}, domain, server, rservice, isIP)
if err != nil {
return
}
c = client.(*Client)
return
}
//PreInit 预初始化服务器连接
func (r *Invoker) PreInit(services ...string) (err error) {
for _, v := range services {
_, err = r.GetClient(v)
if err != nil {
return
}
}
return
}
//Close 关闭当前客户端与服务器的连接
func (r *Invoker) Close() {
r.cache.RemoveIterCb(func(k string, v interface{}) bool {
client := v.(*Client)
client.Close()
return true
})
}