限流(并发控制)

概述

限流器是一种服务治理能力，用于限制服务的并发调用量，以保护服务的稳定性。在本次文档中仅介绍对 rest，grpc 服务的限流器的使用，不介绍限流算法。

限流一般有单节点限流、集群限流（将限流数值对集群节点数求平均值，其本质还是单节点限流）、分布式限流。

本次介绍的是单节点限流，与其说是限流，倒不如说是并发控制更贴切。

准备工作

  1. 创建一个 go mod 的 demo 工程
  1. $ mkdir -p ~/workspace/demo && cd ~/workspace/demo
  2. $ go mod init demo

rest 服务

限流配置

  1. RestConf struct {
  2.     ...
  3.     MaxConns int    `json:",default=10000"` // 最大并发连接数，默认值为 10000 qps
  4.     ... 
  5. }

示例

我们用一个 demo 来介绍一下限流器的使用。

  1. 在 demo 工程中新建目录 rest-limit-demo

    1. $ cd ~/workspace/demo
    2. $ mkdir rest-limit-demo && cd rest-limit-demo

  2. 新建一个 limit.api 文件，将如下内容拷贝到文件中

    limit.api

    1. syntax = "v1"
    3. service limit {
    4.   @handler ping
    5.   get /ping
    6. }

  3. 生成 rest 代码

    1. $ cd ~/workspace/demo/rest-limit-demo
    2. $ goctl api go -api limit.api -dir .

  4. 查看目录结构

    1. $ tree
    2. .
    3. ├── etc
    4. │   └── limit.yaml
    5. ├── internal
    6. │   ├── config
    7. │   │   └── config.go
    8. │   ├── handler
    9. │   │   ├── pinghandler.go
    10. │   │   └── routes.go
    11. │   ├── logic
    12. │   │   └── pinglogic.go
    13. │   ├── svc
    14. │   │   └── servicecontext.go
    15. │   └── types
    16. │       └── types.go
    17. ├── limit.api
    18. └── limit.go

    我们修改一下配置，将 qps 限制为 100，然后逻辑中加一点阻塞逻辑。

  5. 修改配置文件

    将 ~/workspace/demo/rest-limit-demo/etc/limit.yaml 中的 maxConns 修改为 100

  6. 添加逻辑代码

    将 ~/workspace/demo/rest-limit-demo/internal/logic/pinglogic.go 中的 Ping 方法添加阻塞逻辑

    最终代码内容如下

  • limit.yaml
  • pinglogic.go

limit.yaml

  1. Name: limit
  2. Host: 0.0.0.0
  3. Port: 8888
  4. MaxConns: 100
  1. package logic
  3. import (
  4.     "context"
  5.     "time"
  7.     "github.com/zeromicro/go-zero/core/logx"
  9.     "demo/rest-limit-demo/internal/svc"
  10. )
  12. type PingLogic struct {
  13.     logx.Logger
  14.     ctx    context.Context
  15.     svcCtx *svc.ServiceContext
  16. }
  18. func NewPingLogic(ctx context.Context, svcCtx *svc.ServiceContext) *PingLogic {
  19.     return &PingLogic{
  20.         Logger: logx.WithContext(ctx),
  21.         ctx:    ctx,
  22.         svcCtx: svcCtx,
  23.     }
  24. }
  26. func (l *PingLogic) Ping() error {
  27.     time.Sleep(50 * time.Millisecond)
  28.     return nil
  29. }

我们先来运行一下这个最简单的 rest 服务，我们用 hey 工具来简单压测一下接口。

先启动服务

  1. $ cd ~/workspace/demo/rest-limit-demo
  2. $ go run limit.go

单独开个终端压测

  1. # 我们用 hey 工具来进行压测，压测 90 个并发，执行 1 秒
  2. $ hey -z 1s -c 90 -q 1 'http://localhost:8888/ping'
  4. Summary:
  5.   Total:    1.1084 secs
  6.   Slowest:  0.1066 secs
  7.   Fastest:  0.0607 secs
  8.   Average:  0.0890 secs
  9.   Requests/sec: 81.1980
  12. Response time histogram:
  13.   0.061 [1] |■
  14.   0.065 [2] |■■■
  15.   0.070 [8] |■■■■■■■■■■■
  16.   0.074 [13]    |■■■■■■■■■■■■■■■■■
  17.   0.079 [5] |■■■■■■■
  18.   0.084 [0] |
  19.   0.088 [0] |
  20.   0.093 [2] |■■■
  21.   0.097 [23]    |■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
  22.   0.102 [30]    |■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
  23.   0.107 [6] |■■■■■■■■
  26. Latency distribution:
  27.   10% in 0.0682 secs
  28.   25% in 0.0742 secs
  29.   50% in 0.0961 secs
  30.   75% in 0.0983 secs
  31.   90% in 0.0996 secs
  32.   95% in 0.1039 secs
  33.   0% in 0.0000 secs
  35. Details (average, fastest, slowest):
  36.   DNS+dialup:   0.0054 secs, 0.0607 secs, 0.1066 secs
  37.   DNS-lookup:   0.0000 secs, 0.0000 secs, 0.0000 secs
  38.   req write:    0.0002 secs, 0.0000 secs, 0.0011 secs
  39.   resp wait:    0.0832 secs, 0.0576 secs, 0.0942 secs
  40.   resp read:    0.0001 secs, 0.0000 secs, 0.0012 secs
  42. Status code distribution:
  43.   [200] 90 responses

从压测结果来看，90 个请求全部成功，我们来加大并发数，看看会发生什么。

  1. # 我们用 hey 工具来进行压测，压测 110 个并发，执行 1 秒
  2. $ hey -z 1s -c 110 -q 1 'http://127.0.0.1:8888/ping'
  4. Summary:
  5.   Total:    1.0833 secs
  6.   Slowest:  0.0756 secs
  7.   Fastest:  0.0107 secs
  8.   Average:  0.0644 secs
  9.   Requests/sec: 101.5403
  12. Response time histogram:
  13.   0.011 [1] |■
  14.   0.017 [9] |■■■■■■■
  15.   0.024 [0] |
  16.   0.030 [0] |
  17.   0.037 [0] |
  18.   0.043 [0] |
  19.   0.050 [0] |
  20.   0.056 [0] |
  21.   0.063 [2] |■■
  22.   0.069 [45]    |■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
  23.   0.076 [53]    |■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
  26. Latency distribution:
  27.   10% in 0.0612 secs
  28.   25% in 0.0665 secs
  29.   50% in 0.0690 secs
  30.   75% in 0.0726 secs
  31.   90% in 0.0737 secs
  32.   95% in 0.0740 secs
  33.   99% in 0.0756 secs
  35. Details (average, fastest, slowest):
  36.   DNS+dialup:   0.0040 secs, 0.0107 secs, 0.0756 secs
  37.   DNS-lookup:   0.0000 secs, 0.0000 secs, 0.0000 secs
  38.   req write:    0.0001 secs, 0.0000 secs, 0.0025 secs
  39.   resp wait:    0.0602 secs, 0.0064 secs, 0.0741 secs
  40.   resp read:    0.0000 secs, 0.0000 secs, 0.0001 secs
  42. Status code distribution:
  43.   [200] 100 responses
  44.   [503] 10 responses

从压测结果来看，我们的服务只能支持 100 个并发，超过 100 个并发的请求都会被限流，返回 503 状态码。 在服务的日志中也会出现限流相关的错误：

  1. {"@timestamp":"2023-02-09T18:41:55.500+08:00","caller":"internal/log.go:62","content":"(/ping - 127.0.0.1:64163) concurrent connections over 100, rejected with code 503","level":"error","span":"08fa61d49b694e63","trace":"622b2287f32ff2d45f4dfce3eec8e62c"}

grpc 服务

grpc 服务属于内网服务，不对外提供服务，只对内部的其他服务提供服务，所以我们默认情况下不需要对其进行限流。

如果真要对其进行限流，可以通过 grpc 中间件来实现，参考示例如下。

我们用一个最简单的 grpc server 服务来演示一下。

  1. 在 demo 工程中新建目录 grpc-limit-demo

    1. $ cd ~/workspace/demo
    2. $ mkdir grpc-limit-demo && cd grpc-limit-demo

  2. 新建一个 limit.proto 文件，将如下内容拷贝到文件中

    limit.proto

    1. syntax = "proto3";
    3. package proto;
    5. option  go_package = "./proto";
    7. message PingReq{}
    8. message PingResp{}
    10. service limit{
    11.   rpc Ping(PingReq) returns (PingResp);
    12. }

  3. 生成 grpc 代码

    1. $ cd ~/workspace/demo/grpc-limit-demo
    2. $ goctl rpc protoc limit.proto --go_out=.  --go-grpc_out=.  --zrpc_out=.

  4. 查看目录

    1. $ tree
    2. .
    3. ├── etc
    4. │   └── limit.yaml
    5. ├── internal
    6. │   ├── config
    7. │   │   └── config.go
    8. │   ├── logic
    9. │   │   └── pinglogic.go
    10. │   ├── server
    11. │   │   └── limitserver.go
    12. │   └── svc
    13. │       └── servicecontext.go
    14. ├── limit
    15. │   └── limit.go
    16. ├── limit.go
    17. ├── limit.proto
    18. └── proto
    19.     ├── limit.pb.go
    20.     └── limit_grpc.pb.go
    22. 8 directories, 10 files

    我们在 ~/workspace/demo/grpc-limit-demo/internal/logic/limitlogic.go 中实现 Ping 方法，代码如下：

    imitlogic.go

    1. package logic
    3. import (
    4.   "context"
    5.   "time"
    7.   "demo/grpc-limit-demo/internal/svc"
    8.   "demo/grpc-limit-demo/proto"
    10.   "github.com/zeromicro/go-zero/core/logx"
    11. )
    13. type PingLogic struct {
    14.   ctx    context.Context
    15.   svcCtx *svc.ServiceContext
    16.   logx.Logger
    17. }
    19. func NewPingLogic(ctx context.Context, svcCtx *svc.ServiceContext) *PingLogic {
    20.   return &PingLogic{
    21.     ctx:    ctx,
    22.     svcCtx: svcCtx,
    23.     Logger: logx.WithContext(ctx),
    24.   }
    25. }
    27. func (l *PingLogic) Ping(in *proto.PingReq) (*proto.PingResp, error) {
    28.   time.Sleep(50*time.Millisecond)
    30.   return &proto.PingResp{}, nil
    31. }

  5. 在 ~/workspace/demo/grpc-limit-demo/limit.go 中添加中间件：

limit.go

  1. package main
  3. import (
  4.     "context"
  5.     "flag"
  6.     "fmt"
  7.     "net/http"
  9.     "github.com/zeromicro/go-zero/core/logx"
  10.     "github.com/zeromicro/go-zero/core/syncx"
  11.     "google.golang.org/grpc/codes"
  12.     "google.golang.org/grpc/status"
  14.     "demo/grpc-limit-demo/internal/config"
  15.     "demo/grpc-limit-demo/internal/server"
  16.     "demo/grpc-limit-demo/internal/svc"
  17.     "demo/grpc-limit-demo/proto"
  19.     "github.com/zeromicro/go-zero/core/conf"
  20.     "github.com/zeromicro/go-zero/core/service"
  21.     "github.com/zeromicro/go-zero/zrpc"
  22.     "google.golang.org/grpc"
  23.     "google.golang.org/grpc/reflection"
  24. )
  26. var configFile = flag.String("f", "etc/limit.yaml", "the config file")
  28. func main() {
  29.     flag.Parse()
  31.     var c config.Config
  32.     conf.MustLoad(*configFile, &c)
  33.     ctx := svc.NewServiceContext(c)
  35.     s := zrpc.MustNewServer(c.RpcServerConf, func(grpcServer *grpc.Server) {
  36.         proto.RegisterLimitServer(grpcServer, server.NewLimitServer(ctx))
  38.         if c.Mode == service.DevMode || c.Mode == service.TestMode {
  39.             reflection.Register(grpcServer)
  40.         }
  41.     })
  42.     var n = 100
  43.     l := syncx.NewLimit(n)
  44.     s.AddUnaryInterceptors(func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
  45.         if l.TryBorrow() {
  46.             defer func() {
  47.                 if err := l.Return(); err != nil {
  48.                     logx.Error(err)
  49.                 }
  50.             }()
  51.             return handler(ctx, req)
  52.         } else {
  53.             logx.Errorf("concurrent connections over %d, rejected with code %d",
  54.                 n, http.StatusServiceUnavailable)
  55.             return nil, status.Error(codes.Unavailable, "concurrent connections over limit")
  56.         }
  57.     })
  58.     defer s.Stop()
  60.     fmt.Printf("Starting rpc server at %s...\n", c.ListenOn)
  61.     s.Start()
  62. }
  64. }

我们将生成的配置文件 ~/workspace/demo/grpc-limit-demo/etc/limit.yaml 文件中的 etcd 配置删除掉，用直连的方式来启动 grpc server。

limit.yaml

  1. Name: limit.rpc
  2. ListenOn: 0.0.0.0:8080

启动服务

  1. $ cd ~/workspace/demo/grpc-limit-demo
  2. $ go run limit.go

现在 grpc server 服务有了，我们用 ghz 来压测一下 。

  1. $ cd ~/workspace/demo/grpc-limit-demo
  2. # 压测 90 qps，共请求 110 次
  3. $ ghz --insecure --proto=limit.proto --call=proto.limit.Ping -d '{}' -c 90 -n 110  127.0.0.1:8080 
  5. Summary:
  6.   Count:        110
  7.   Total:        108.94 ms
  8.   Slowest:      60.51 ms
  9.   Fastest:      50.24 ms
  10.   Average:      55.73 ms
  11.   Requests/sec: 1009.75
  13. Response time histogram:
  14.   50.240 [1]  |∎∎
  15.   51.266 [14] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  16.   52.293 [9]  |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  17.   53.320 [3]  |∎∎∎∎∎∎
  18.   54.347 [9]  |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  19.   55.374 [5]  |∎∎∎∎∎∎∎∎∎∎
  20.   56.401 [9]  |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  21.   57.427 [20] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  22.   58.454 [19] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  23.   59.481 [14] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  24.   60.508 [7]  |∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  26. Latency distribution:
  27.   10 % in 50.88 ms 
  28.   25 % in 53.40 ms 
  29.   50 % in 56.81 ms 
  30.   75 % in 57.96 ms 
  31.   90 % in 58.93 ms 
  32.   95 % in 59.52 ms 
  33.   99 % in 60.26 ms 
  35. Status code distribution:
  36.   [OK]   110 responses

可以看到所有的请求都是成功的，我们再来压测 110 qps。

  1. $ cd ~/workspace/demo/grpc-limit-demo
  2. # 压测 110 qps，共请求 110 次
  3. $ ghz --insecure --proto=limit.proto --call=proto.limit.Ping -d '{}' -c 110 -n 110  127.0.0.1:8080
  5. Summary:
  6.   Count:        110
  7.   Total:        68.35 ms
  8.   Slowest:      67.53 ms
  9.   Fastest:      58.76 ms
  10.   Average:      58.03 ms
  11.   Requests/sec: 1609.33
  13. Response time histogram:
  14.   58.763 [1]  |∎∎
  15.   59.640 [7]  |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  16.   60.517 [12] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  17.   61.394 [7]  |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  18.   62.272 [10] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  19.   63.149 [15] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  20.   64.026 [14] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  21.   64.903 [19] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  22.   65.780 [9]  |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  23.   66.657 [3]  |∎∎∎∎∎∎
  24.   67.534 [3]  |∎∎∎∎∎∎
  26. Latency distribution:
  27.   10 % in 59.81 ms 
  28.   25 % in 60.96 ms 
  29.   50 % in 62.96 ms 
  30.   75 % in 64.52 ms 
  31.   90 % in 65.29 ms 
  32.   95 % in 66.09 ms 
  33.   99 % in 67.03 ms 
  35. Status code distribution:
  36.   [Unavailable]   10 responses    
  37.   [OK]            100 responses   
  39. Error distribution:
  40.   [10]   rpc error: code = Unavailable desc = concurrent connections over limit

可以看到，当并发量超过 100 时，就会返回 rpc error: code = Unavailable desc = concurrent connections over limit。

参考文献

  1. 《轻量级压测工具 hey》
  2. 《grpc 压测工作 ghz》