如何用didip/tollbooth实现全局并发与IP维度请求限流？

背景 #

我是限流技术新手，想用tollbooth库实现HTTP请求限流，也了解过令牌桶算法的相关内容。现在要给一个测试应用实现两个限流规则：

• 全局限制最大并发请求数为10（不区分请求IP）
• 基于请求IP设置最大突发请求量为3

以下是我参考tollbooth示例编写的初始代码：

package main

import (
    "net/http"
    "time"

    "github.com/didip/tollbooth/v7"
    "github.com/didip/tollbooth/v7/limiter"
)

func main() {
    lmt := tollbooth.NewLimiter(3, &limiter.ExpirableOptions{DefaultExpirationTTL: time.Hour})

    http.Handle("/", tollbooth.LimitFuncHandler(lmt, HelloHandler))
    http.ListenAndServe(":8080", nil)
}

func HelloHandler(w http.ResponseWriter, req *http.Request) {
    w.Write([]byte("Hello, World!"))
}

测试时连续执行curl -i localhost:8080，超过阈值会收到HTTP/1.1 429 Too Many Requests错误，这部分符合预期，但还有两个疑问：

问题 #

1. 如何用tollbooth实现全局最大并发请求数限制（比如10）？这么做是否有意义？我认为有必要，因为仅做IP限流的话，大量不同IP同时访问可能导致服务器内存耗尽。
2. 我当前的限流实现是否正确？有没有遗漏的点？这类场景是不是更适合用云环境的负载均衡器来处理？

更新：可运行的完整代码 #

基于回答编写的兼顾两种限流规则的可运行代码：

package main

import (
    "net/http"
    "sync"
    "time"

    "github.com/didip/tollbooth/v7"
    "github.com/didip/tollbooth/v7/limiter"
)

func main() {
    ipLimiter := tollbooth.NewLimiter(3, &limiter.ExpirableOptions{DefaultExpirationTTL: time.Hour})
    globalLimiter := NewConcurrentLimiter(10)

    http.Handle("/", globalLimiter.LimitConcurrentRequests(ipLimiter, HelloHandler))
    http.ListenAndServe(":8080", nil)
}

func HelloHandler(w http.ResponseWriter, req *http.Request) {
    w.Write([]byte("Hello, World!"))
}

type ConcurrentLimiter struct {
    max     int
    current int
    mut     sync.Mutex
}

func NewConcurrentLimiter(limit int) *ConcurrentLimiter {
    return &ConcurrentLimiter{
        max: limit,
    }
}

func (limiter *ConcurrentLimiter) LimitConcurrentRequests(lmt *limiter.Limiter,
    handler func(http.ResponseWriter, *http.Request)) http.Handler {

    middle := func(w http.ResponseWriter, r *http.Request) {

        limiter.mut.Lock()
        maxHit := limiter.current == limiter.max

        if maxHit {
            limiter.mut.Unlock()
            http.Error(w, http.StatusText(429), http.StatusTooManyRequests)
            return
        }

        limiter.current += 1
        limiter.mut.Unlock()

        defer func() {
            limiter.mut.Lock()
            limiter.current -= 1
            limiter.mut.Unlock()
        }()

        // 无限流错误，继续处理请求
        handler(w, r)
    }

    return tollbooth.LimitHandler(lmt, http.HandlerFunc(middle))
}

内容的提问来源于stack exchange，提问作者Floating Sunfish