基于go-micro微服务的实战-Gateway网关层的限流降级(八)

基于go-micro微服务的实战-Gateway网关层的限流降级(八)

---

文章最后附带完整代码

这一节主要是在Gateway网关层，基于go-micro的装饰器引入限流和降级。限流降级用的是开源库hystrix,类似java的hystrix，这里不做具体介绍和使用，可自行查看文档。

设计流程是这样

1. 请求到达网关层，超过一定并发数则限制请求，直接返回默认结果
2. 请求到达网关层，转发到服务，服务返回错误信息个数达到一定数量，接口熔断，一定时间内请求直接返回默认结果

---

第一步：网关层新增装饰器

在grpc_gateway中新增wrapper目录和client_wrapper.go客户端请求装饰器

import "github.com/asim/go-micro/v3/client"
...
type CliWrapper struct {
	client.Client
}

func (c *CliWrapper) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error{
	log.Println("client wrapper!!!")
	//限流唯一名
	commandName := req.Service() + "." + req.Endpoint()

	//行为主体
	action := func() error {
		return c.Client.Call(ctx, req, rsp, opts...)
	}

    //限流和降级
	if HystrixLimit(commandName, req.Endpoint(), action) == nil {
		return nil
	}else{
        //限流降级默认返回结果处理
		HystrixFallback(rsp)
		return nil
	}
}

func NewCliWrapper(c client.Client) client.Client{
	return &CliWrapper{c}
}
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

第二步：通过hystrix库实现限流和降级

在wrapper目录下新增client_limit.go限流处理器。
*注意，这里的限流的并发指的是同时并发，如果串行，单进程一个for循环1w次也是只有一个并发。

几个参数介绍

限流的两个参数：
Timeout：时间毫秒，指定时间内为一个窗口
MaxConcurrentRequests：在窗口时间内，并发最大数的限制

下面三个参数是熔断参数，也就是有RequestVolumeThreshold个请求，就判断超过ErrorPercentThreshold比例则熔断，熔断时间SleepWindow
ErrorPercentThreshold：错误率，百分比，超过设定值则打开熔断器，默认50(50%)
RequestVolumeThreshold：请求阈值  熔断器是否打开首先要满足这个条件；这里的设置表示至少有2个请求才进行ErrorPercentThreshold错误百分比计算，默认20
SleepWindow：过多长时间，熔断器再次检测是否开启（是否启动服务运行）。单位毫秒，默认5秒
1
2
3
4
5
6
7
8

具体的限流处理器代码

import "github.com/afex/hystrix-go/hystrix"

//限流处理
func HystrixLimit(commandName, endpoint string, action func() error) error{

	var config hystrix.CommandConfig
	switch endpoint {
	case "UserService.UserLogin":			  //登录接口，限制3秒内只能有2个并发
		config = hystrix.CommandConfig{
			Timeout: 3000,
			MaxConcurrentRequests: 2,

			//下面三个参数是熔断配置，也就是有RequestVolumeThreshold个请求，就判断超过ErrorPercentThreshold比例则熔断，熔断时间SleepWindow
			ErrorPercentThreshold: 25,        //错误率，百分比，超过设定值则打开熔断器，默认50(50%)
			RequestVolumeThreshold:2,         //请求阈值  熔断器是否打开首先要满足这个条件；这里的设置表示至少有2个请求才进行ErrorPercentThreshold错误百分比计算，默认20
			SleepWindow: 5000,                //过多长时间，熔断器再次检测是否开启（是否启动服务运行）。单位毫秒，默认5秒
		}
	case "UserService.UserReg":               //主从接口，限制5秒内只能有10个并发
		config = hystrix.CommandConfig{
			Timeout: 5000,
			MaxConcurrentRequests: 10,
			ErrorPercentThreshold: 25,
			RequestVolumeThreshold:10,
			SleepWindow: 2000,
		}
	default:                                 //默认其它非指定接口的处理
		config = hystrix.CommandConfig{
			Timeout: 5000,
			MaxConcurrentRequests: 10000,
			ErrorPercentThreshold: 25,
			RequestVolumeThreshold:100,
			SleepWindow: 5000,
		}
	}
	hystrix.ConfigureCommand(commandName, config)

	return hystrix.Do(commandName, func() error{
		return action()
	}, func(err error) error{
		return err
	})
}

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

第三步：限流降级中请求默认结果的处理

主要就是针对返回结果，对返回结果做一个默认内容的处理，友好提示等等，根据实际场景做相应处理。

func HystrixFallback(resp interface{}) {
	switch t:= resp.(type) {
	case *pb.RegResp:
		t.Status = 2
		t.Msg = "please wait to try"
	case *pb.LoginResp:
		t.Status = 2
		t.Msg = "please wait to try"
	case *pb.TestResp:
		t.Msg = "is limit"
	default:

	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

第四步：测试验证

限流测试：
模拟发起登录请求测试，发送6个

	body := "{\"phone\":\"15222222222\",\"pwd\":\"123789\"}"
	//post 方法参数，第一个参数为请求url,第二个参数 是contentType, 第三个参数为请求体[]byte格式
	w := sync.WaitGroup{}
	for i := 1; i <= 6; i++{
		w.Add(1)
		go func() {
			responce, err := http.Post("http://localhost:55001/user/login", "application/json", bytes.NewBuffer([]byte(body)))
			if err != nil {
				fmt.Println("net http post method err,", err)
			}
			defer responce.Body.Close()
			w.Done()
		}()
	}
	w.Wait()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

按上面第2步的配置，登录接口3秒内限制2个，实际结果6个请求，只有2个可以通过，4个直接返回，符合预期！如图。

降级测试：

修改用户服务的登录接口，直接弄个暴力报错

func (u *UserHandler) UserLogin(ctx context.Context, req *pb.LoginReq, resp *pb.LoginResp) error{
	var i int = 0
	log.Println(10/i)
1
2
3

模拟发起登录请求测试，每次发送4个，睡眠3秒

	w := sync.WaitGroup{}
	for i := 0; i <= 3; i++{
		w.Add(1)
		go func() {
			responce, err := http.Post("http://localhost:55001/user/login", "application/json", bytes.NewBuffer([]byte(body)))
			if err != nil {
				fmt.Println("net http post method err,", err)
			}
			defer responce.Body.Close()
			w.Done()
		}()
	}
	time.Sleep(3 *time.Second)
	for i := 0; i <= 3; i++{
		w.Add(1)
		go func() {
			responce, err := http.Post("http://localhost:55001/user/login", "application/json", bytes.NewBuffer([]byte(body)))
			if err != nil {
				fmt.Println("net http post method err,", err)
			}
			defer responce.Body.Close()
			w.Done()
		}()
	}
	w.Wait()
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

如下图，3秒内只有2个请求可以通过，但是后续的6个请求，接口熔断，不发起请求，直接走默认返回结果

gitee完整代码链接