How does Go implement the HTTP request throttling example

When developing high concurrency systems, use three tools to protect the system: caching, downgrading, and throttling! In order to ensure the flexibility and stability of the online system in the peak period of business, the most effective solution is to degrade the service, and current limiting is one of the most commonly used solutions in the degraded system.

Here to recommend a open source library https: / / github com didip/tollbooth but, if you want to be a simple, lightweight, or just want to learn something, realizes own middleware to deal with the speed limit is not difficult. Today we are going to talk about how to implement your own stream-limiting middleware

First we need to install a dependency package that provides Token bucket (token bucket algorithm) on which the implementation of toolbooth mentioned above is based

$ go get golang.org/x/time/rate

Well, let's look at the implementation of the Demo code first:

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"))
}

Let's take a look at the source code of ES27en. 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:

With an average user-configured send rate of r, one token is added to the bucket every 1/r second (r tokens are added to the bucket per second) and a maximum of b tokens can be stored in the bucket. If the token arrives when the token bucket is full, the token is discarded;

Achieve user granularity of current limiting

While it is useful to use a single global rate limiter in some cases, another common case is to enforce the rate limiter for each user based on identifiers such as the IP address or API key. 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 just a simple implementation, but there are a lot of things to consider if we want to implement streaming throttling in API-ES49en for microservices. Suggest you guys can see https: / / github com/didip tollbooth source.