There are three tools you can use to protect a high-concurrency system: caching, degradation, and limiting traffic! In order to ensure the flexibility and stability of the online system during the peak hours, the most effective scheme is service degradation, and traffic limiting is one of the schemes most often adopted by the degraded system.
Here is a recommended open source library github.com/didip/tollb… However, if you want something simple, lightweight, or just want to learn, it’s not difficult to implement your own middleware to handle rate limits. Today we are going to talk about how to achieve their own a limiting middleware
The first step is to install a dependency package that provides the Token bucket (Token bucket algorithm) on which the toolBooth implementation is based
- $ go get golang.org/x/time/rate
Ok, let’s look at the implementation of the Demo code:
limit.go
- package main
- import (
- “net/http”
- “golang.org/x/time/rate”
- )
- var limiter = rate.NewLimiter(2, 5)
- func limit(next http.Handler) http.Handler {
- return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request ) {
- if limiter.Allow() == false {
- http.Error(w, http.StatusText(429 ), http.StatusTooManyRequests)
- return
- }
- next.ServeHTTP(w, r)
- })
- }
main.go
- package main
- import (
- “net/http”
- )
- func main() {
- mux := http.NewServeMux()
- mux.HandleFunc(“/”, okHandler)
- // Wrap the servemux with the limit middleware.
- http.ListenAndServe(“:4000”, limit( mux))
- }
- func okHandler(w http.ResponseWriter, r *http.Request) {
- w.Write([]byte(“OK” ))
- }
Rate.newlimiter (rate.newlimiter)
- // Copyright 2015 The Go Authors. All rights reserved.
- // Use of this source code is governed by a BSD-style
- // license that can be found in the LICENSE file.
- // Package rate provides a rate limiter.
- package rate
- import (
- “fmt”
- “math”
- “sync”
- “time”
- “golang.org/x/net/context”
- )
- // Limit defines the maximum frequency of some events.
- // Limit is represented as number of events per second.
- // A zero Limit allows no events.
- type Limit float64
- // Inf is the infinite rate limit; it allows all events (even if burst is zero).
- const Inf = Limit(math.MaxFloat64)
- // Every converts a minimum time interval between events to a Limit.
- func Every(interval time.Duration) Limit {
- if interval <= 0 {
- return Inf
- }
- return 1 / Limit(interval.Seconds ())
- }
- // A Limiter controls how frequently events are allowed to happen.
- // It implements a “token bucket” of size b, initially full and refilled
- // at rate r tokens per second.
- // Informally, in any large enough time interval, the Limiter limits the
- // rate to r tokens per second, with a maximum burst size of b events.
- // As a special case, if r == Inf (the infinite rate), b is ignored.
- // See https://en.wikipedia.org/wiki/Token_bucket for more about token buckets.
- //
- // The zero value is a valid Limiter, but it will reject all events.
- // Use NewLimiter to create non-zero Limiters.
- //
- // Limiter has three main methods, Allow, Reserve, and Wait.
- // Most callers should use Wait.
- //
- // Each of the three methods consumes a single token.
- // They differ in their behavior when no token is available.
- // If no token is available, Allow returns false.
- // If no token is available, Reserve returns a reservation for a future token
- // and the amount of time the caller must wait before using it.
- // If no token is available, Wait blocks until one can be obtained
- // or its associated context.Context is canceled.
- //
- // The methods AllowN, ReserveN, and WaitN consume n tokens.
- type Limiter struct {
- limit Limit
- burst int
- mu sync.Mutex
- tokens float64
- // last is the last time the limiter’s tokens field was updated
- last time.Time
- // lastEvent is the latest time of a rate-limited event (past or future)
- lastEvent time.Time
- }
- // Limit returns the maximum overall event rate.
- func (lim *Limiter) Limit () Limit {
- lim.mu.Lock()
- defer lim.mu.Unlock()
- return lim.limit
- }
- // Burst returns the maximum burst size. Burst is the maximum number of tokens
- // that can be consumed in a single call to Allow, Reserve, or Wait, so higher
- // Burst values allow more events to happen at once.
- // A zero Burst allows no events, unless limit == Inf.
- func (lim *Limiter) Burst () int {
- return lim.burst
- }
- // NewLimiter returns a new Limiter that allows events up to rate r and permits
- // bursts of at most b tokens.
- func NewLimiter(r Limit, b int) *Limiter {
- return &Limiter{
- limit: r,
- burst: b,
- }
- }
- // Allow is shorthand for AllowN(time.Now(), 1).
- func (lim *Limiter) Allow () bool {
- return lim.AllowN(time.Now() , 1)
- }
- // AllowN reports whether n events may happen at time now.
- // Use this method if you intend to drop / skip events that exceed the rate limit.
- // Otherwise use Reserve or Wait.
- func (lim *Limiter) AllowN (now time.Time, n int) bool {
- return lim.reserveN(now, n, 0).ok
- }
- // A Reservation holds information about events that are permitted by a Limiter to happen after a delay.
- // A Reservation may be canceled, which may enable the Limiter to permit additional events.
- type Reservation struct {
- ok bool
- lim *Limiter
- tokens int
- timeToAct time.Time
- // This is the Limit at reservation time, it can change later.
- limit Limit
- }
- // OK returns whether the limiter can provide the requested number of tokens
- // within the maximum wait time. If OK is false, Delay returns InfDuration, and
- // Cancel does nothing.
- func (r *Reservation) OK () bool {
- return r.ok
- }
- // Delay is shorthand for DelayFrom(time.Now()).
- func (r *Reservation) Delay () time.Duration {
- return r.DelayFrom(time.Now() )
- }
- // InfDuration is the duration returned by Delay when a Reservation is not OK.
- const InfDuration = time.Duration(1<<63 – 1)
- // DelayFrom returns the duration for which the reservation holder must wait
- // before taking the reserved action. Zero duration means act immediately.
- // InfDuration means the limiter cannot grant the tokens requested in this
- // Reservation within the maximum wait time.
- func (r *Reservation) DelayFrom (now time.Time) time.Duration {
- if ! r.ok {
- return InfDuration
- }
- delay := r.timeToAct.Sub(now)
- if delay < 0 {
- return 0
- }
- return delay
- }
- // Cancel is shorthand for CancelAt(time.Now()).
- func (r *Reservation) Cancel () {
- r.CancelAt(time.Now())
- return
- }
- // CancelAt indicates that the reservation holder will not perform the reserved action
- // and reverses the effects of this Reservation on the rate limit as much as possible,
- // considering that other reservations may have already been made.
- func (r *Reservation) CancelAt (now time.Time) {
- if ! r.ok {
- return
- }
- r.lim.mu.Lock()
- defer r.lim.mu.Unlock()
- if r.lim.limit == Inf || r.tokens == 0 || r.timeToAct.Before(now ) {
- return
- }
- // calculate tokens to restore
- // The duration between lim.lastEvent and r.timeToAct tells us how many tokens were reserved
- // after r was obtained. These tokens should not be restored.
- restoreTokens := float64(r.tokens) – r.limit.tokensFromDuration (r.lim.lastEvent.Sub(r.timeToAct))
- if restoreTokens <= 0 {
- return
- }
- // advance time to now
- now, _, tokens := r.lim.advance(now)
- // calculate new number of tokens
- tokens += restoreTokens
- if burst := float64(r.lim.burst); tokens > burst {
- tokens = burst
- }
- // update state
- r.lim.last = now
- r.lim.tokens = tokens
- if r.timeToAct == r.lim.lastEvent {
- prevEvent := r.timeToAct.Add(r.limit.durationFromTokens( float64(-r.tokens)))
- if ! prevEvent.Before(now) {
- r.lim.lastEvent = prevEvent
- }
- }
- return
- }
- // Reserve is shorthand for ReserveN(time.Now(), 1).
- func (lim *Limiter) Reserve () *Reservation {
- return lim.ReserveN(time.Now() , 1)
- }
- // ReserveN returns a Reservation that indicates how long the caller must wait before n events happen.
- // The Limiter takes this Reservation into account when allowing future events.
- // ReserveN returns false if n exceeds the Limiter’s burst size.
- // Usage example:
- // r, ok := lim.ReserveN(time.Now(), 1)
- // if ! ok {
- // // Not allowed to act! Did you remember to set lim.burst to be > 0 ?
- // }
- // time.Sleep(r.Delay())
- // Act()
- // Use this method if you wish to wait and slow down in accordance with the rate limit without dropping events.
- // If you need to respect a deadline or cancel the delay, use Wait instead.
- // To drop or skip events exceeding rate limit, use Allow instead.
- func (lim *Limiter) ReserveN (now time.Time, n int) *Reservation {
- r := lim.reserveN(now, n, InfDuration)
- return &r
- }
- // Wait is shorthand for WaitN(ctx, 1).
- func (lim *Limiter) Wait (ctx context.Context) (err error) {
- return lim.WaitN(ctx, 1)
- }
- // WaitN blocks until lim permits n events to happen.
- // It returns an error if n exceeds the Limiter’s burst size, the Context is
- // canceled, or the expected wait time exceeds the Context’s Deadline.
- func (lim *Limiter) WaitN (ctx context.Context, n int) (err error) {
- if n > lim.burst {
- return fmt.Errorf(“rate: Wait(n=%d) exceeds limiter’s burst %d”, n, lim.burst )
- }
- // Check if ctx is already cancelled
- select {
- case <-ctx.Done():
- return ctx.Err()
- default:
- }
- // Determine wait limit
- now := time.Now()
- waitLimit := InfDuration
- if deadline, ok := ctx.Deadline(); ok {
- waitLimit = deadline.Sub(now)
- }
- // Reserve
- r := lim.reserveN(now, n, waitLimit)
- if ! r.ok {
- return fmt.Errorf(“rate: Wait(n=%d) would exceed context deadline”, n )
- }
- // Wait
- t := time.NewTimer(r.DelayFrom(now ))
- defer t.Stop()
- select {
- case <-t.C:
- // We can proceed.
- return nil
- case <-ctx.Done():
- // Context was canceled before we could proceed. Cancel the
- // reservation, which may permit other events to proceed sooner.
- r.Cancel()
- return ctx.Err()
- }
- }
- // SetLimit is shorthand for SetLimitAt(time.Now(), newLimit).
- func (lim *Limiter) SetLimit (newLimit Limit) {
- lim.SetLimitAt(time.Now() , newLimit)
- }
- // SetLimitAt sets a new Limit for the limiter. The new Limit, and Burst, may be violated
- // or underutilized by those which reserved (using Reserve or Wait) but did not yet act
- // before SetLimitAt was called.
- func (lim *Limiter) SetLimitAt (now time.Time, newLimit Limit) {
- lim.mu.Lock()
- defer lim.mu.Unlock()
- now, _, tokens := lim.advance(now)
- lim.last = now
- lim.tokens = tokens
- lim.limit = newLimit
- }
- // reserveN is a helper method for AllowN, ReserveN, and WaitN.
- // maxFutureReserve specifies the maximum reservation wait duration allowed.
- // reserveN returns Reservation, not *Reservation, to avoid allocation in AllowN and WaitN.
- func (lim *Limiter) reserveN (now time.Time, n int, maxFutureReserve time.Duration) Reservation {
- lim.mu.Lock()
- defer lim.mu.Unlock()
- if lim.limit == Inf {
- return Reservation{
- ok: true,
- lim: lim,
- tokens: n,
- timeToAct: now,
- }
- }
- now, last, tokens := lim.advance(now)
- // Calculate the remaining number of tokens resulting from the request.
- tokens -= float64(n)
- // Calculate the wait duration
- var waitDuration time.Duration
- if tokens < 0 {
- waitDuration = lim.limit.durationFromTokens(-tokens)
- }
- // Decide result
- ok := n <= lim.burst && waitDuration <= maxFutureReserve
- // Prepare reservation
- r := Reservation{
- ok: ok,
- lim: lim,
- limit: lim.limit,
- }
- if ok {
- r.tokens = n
- r.timeToAct = now.Add(waitDuration)
- }
- // Update state
- if ok {
- lim.last = now
- lim.tokens = tokens
- lim.lastEvent = r.timeToAct
- } else {
- lim.last = last
- }
- return r
- }
- // advance calculates and returns an updated state for lim resulting from the passage of time.
- // lim is not changed.
- func (lim *Limiter) advance(now time.Time) (newNow time.Time, newLast time.Time, newTokens float64 ) {
- last := lim.last
- if now.Before(last) {
- last = now
- }
- // Avoid making delta overflow below when last is very old.
- maxElapsed := lim.limit.durationFromTokens(float64( lim.burst) – lim.tokens)
- elapsed := now.Sub(last)
- if elapsed > maxElapsed {
- elapsed = maxElapsed
- }
- // Calculate the new number of tokens, due to time that passed.
- delta := lim.limit.tokensFromDuration(elapsed)
- tokens := lim.tokens + delta
- if burst := float64(lim.burst); tokens > burst {
- tokens = burst
- }
- return now, last, tokens
- }
- // durationFromTokens is a unit conversion function from the number of tokens to the duration
- // of time it takes to accumulate them at a rate of limit tokens per second.
- func (limit Limit) durationFromTokens(tokens float64) time.Duration {
- seconds := tokens / float64(limit)
- return time.Nanosecond * time.Duration(1e9*seconds)
- }
- // tokensFromDuration is a unit conversion function from a time duration to the number of tokens
- // which could be accumulated during that duration at a rate of limit tokens per second.
- func (limit Limit) tokensFromDuration(d time.Duration) float64 {
- return d.Seconds() * float64 (limit)
- }
Algorithm description: If the average sending rate is set to R, one token is added to the bucket every 1/r second (r tokens are added to the bucket every second). A bucket can store a maximum of B tokens. If the bucket is full when the token arrives, the token is discarded;
Implement traffic limiting of user granularity
While using a single global rate limiter can be useful in some cases, it is also common to enforce rate limiters for each user based on identifiers such as IP addresses or API keys. We will use the IP address as the identifier. The simple implementation code is as follows:
- package main
- import (
- “net/http”
- “sync”
- “time”
- “golang.org/x/time/rate”
- )
- // Create a custom visitor struct which holds the rate limiter for each
- // visitor and the last time that the visitor was seen.
- type visitor struct {
- limiter *rate.Limiter
- lastSeen time.Time
- }
- // Change the the map to hold values of the type visitor.
- var visitors = make(map[string] *visitor)
- var mtx sync.Mutex
- // Run a background goroutine to remove old entries from the visitors map.
- func init() {
- go cleanupVisitors()
- }
- func addVisitor(ip string) *rate.Limiter {
- limiter := rate.NewLimiter(2, 5)
- mtx.Lock()
- // Include the current time when creating a new visitor.
- visitors[ip] = &visitor{limiter, time. Now()}
- mtx.Unlock()
- return limiter
- }
- func getVisitor(ip string) *rate.Limiter {
- mtx.Lock()
- v, exists := visitors[ip]
- if ! exists {
- mtx.Unlock()
- return addVisitor(ip)
- }
- // Update the last seen time for the visitor.
- v.lastSeen = time.Now()
- mtx.Unlock()
- return v.limiter
- }
- // Every minute check the map for visitors that haven’t been seen for
- // more than 3 minutes and delete the entries.
- func cleanupVisitors() {
- for {
- time.Sleep(time.Minute)
- mtx.Lock()
- for ip, v := range visitors {
- if time.Now().Sub( v.lastSeen) > 3*time.Minute {
- delete(visitors, ip)
- }
- }
- mtx.Unlock()
- }
- }
- func limit(next http.Handler) http.Handler {
- return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request ) {
- limiter := getVisitor(r.RemoteAddr)
- if limiter.Allow() == false {
- http.Error(w, http.StatusText(429 ), http.StatusTooManyRequests)
- return
- }
- next.ServeHTTP(w, r)
- })
- }
Of course, this is a simple implementation, but there are a lot of things to consider if we want to implement limiting in apI-Gateway for microservices. I suggest you check out github.com/didip/tollb… The source code.
Your support will encourage us to continue to create!
WeChat pay
Alipay
Scan the QR code with wechat to tip
Scan the QR code with Alipay
WeChat
Sina Weibo
Qzone
Evernote
Facebook
Twitter
Email
Telegram
share