How does Go implement TRAFFIC limiting for HTTP requests

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

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

Ok, let’s look at the implementation of the Demo code:

limit.go

1. package main
3. import (
4. “net/http”
6. “golang.org/x/time/rate”
7. )
9. var limiter = rate.NewLimiter(2, 5)
11. func limit(next http.Handler) http.Handler {
12. return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request ) {
13. if limiter.Allow() == false {
14. http.Error(w, http.StatusText(429 ), http.StatusTooManyRequests)
15. return
16. }
18. next.ServeHTTP(w, r)
19. })
20. }

Rate.newlimiter (rate.newlimiter)

1. // Copyright 2015 The Go Authors. All rights reserved.
2. // Use of this source code is governed by a BSD-style
3. // license that can be found in the LICENSE file.
5. // Package rate provides a rate limiter.
6. package rate
8. import (
9. “fmt”
10. “math”
11. “sync”
12. “time”
14. “golang.org/x/net/context”
15. )
17. // Limit defines the maximum frequency of some events.
18. // Limit is represented as number of events per second.
19. // A zero Limit allows no events.
20. type Limit float64
22. // Inf is the infinite rate limit; it allows all events (even if burst is zero).
23. const Inf = Limit(math.MaxFloat64)
25. // Every converts a minimum time interval between events to a Limit.
26. func Every(interval time.Duration) Limit {
27. if interval <= 0 {
28. return Inf
29. }
30. return 1 / Limit(interval.Seconds ())
31. }
33. // A Limiter controls how frequently events are allowed to happen.
34. // It implements a “token bucket” of size b, initially full and refilled
35. // at rate r tokens per second.
36. // Informally, in any large enough time interval, the Limiter limits the
37. // rate to r tokens per second, with a maximum burst size of b events.
38. // As a special case, if r == Inf (the infinite rate), b is ignored.
39. // See https://en.wikipedia.org/wiki/Token_bucket for more about token buckets.
40. //
41. // The zero value is a valid Limiter, but it will reject all events.
42. // Use NewLimiter to create non-zero Limiters.
43. //
44. // Limiter has three main methods, Allow, Reserve, and Wait.
45. // Most callers should use Wait.
46. //
47. // Each of the three methods consumes a single token.
48. // They differ in their behavior when no token is available.
49. // If no token is available, Allow returns false.
50. // If no token is available, Reserve returns a reservation for a future token
51. // and the amount of time the caller must wait before using it.
52. // If no token is available, Wait blocks until one can be obtained
53. // or its associated context.Context is canceled.
54. //
55. // The methods AllowN, ReserveN, and WaitN consume n tokens.
56. type Limiter struct {
57. limit Limit
58. burst int
60. mu sync.Mutex
61. tokens float64
62. // last is the last time the limiter’s tokens field was updated
63. last time.Time
64. // lastEvent is the latest time of a rate-limited event (past or future)
65. lastEvent time.Time
66. }
68. // Limit returns the maximum overall event rate.
69. func (lim *Limiter) Limit () Limit {
70. lim.mu.Lock()
71. defer lim.mu.Unlock()
72. return lim.limit
73. }
75. // Burst returns the maximum burst size. Burst is the maximum number of tokens
76. // that can be consumed in a single call to Allow, Reserve, or Wait, so higher
77. // Burst values allow more events to happen at once.
78. // A zero Burst allows no events, unless limit == Inf.
79. func (lim *Limiter) Burst () int {
80. return lim.burst
81. }
83. // NewLimiter returns a new Limiter that allows events up to rate r and permits
84. // bursts of at most b tokens.
85. func NewLimiter(r Limit, b int) *Limiter {
86. return &Limiter{
87. limit: r,
88. burst: b,
89. }
90. }
92. // Allow is shorthand for AllowN(time.Now(), 1).
93. func (lim *Limiter) Allow () bool {
94. return lim.AllowN(time.Now() , 1)
95. }
97. // AllowN reports whether n events may happen at time now.
98. // Use this method if you intend to drop / skip events that exceed the rate limit.
99. // Otherwise use Reserve or Wait.
100. func (lim *Limiter) AllowN (now time.Time, n int) bool {
101. return lim.reserveN(now, n, 0).ok
102. }
104. // A Reservation holds information about events that are permitted by a Limiter to happen after a delay.
105. // A Reservation may be canceled, which may enable the Limiter to permit additional events.
106. type Reservation struct {
107. ok bool
108. lim *Limiter
109. tokens int
110. timeToAct time.Time
111. // This is the Limit at reservation time, it can change later.
112. limit Limit
113. }
115. // OK returns whether the limiter can provide the requested number of tokens
116. // within the maximum wait time. If OK is false, Delay returns InfDuration, and
117. // Cancel does nothing.
118. func (r *Reservation) OK () bool {
119. return r.ok
120. }
122. // Delay is shorthand for DelayFrom(time.Now()).
123. func (r *Reservation) Delay () time.Duration {
124. return r.DelayFrom(time.Now() )
125. }
127. // InfDuration is the duration returned by Delay when a Reservation is not OK.
128. const InfDuration = time.Duration(1<<63 – 1)
130. // DelayFrom returns the duration for which the reservation holder must wait
131. // before taking the reserved action. Zero duration means act immediately.
132. // InfDuration means the limiter cannot grant the tokens requested in this
133. // Reservation within the maximum wait time.
134. func (r *Reservation) DelayFrom (now time.Time) time.Duration {
135. if ! r.ok {
136. return InfDuration
137. }
138. delay := r.timeToAct.Sub(now)
139. if delay < 0 {
140. return 0
141. }
142. return delay
143. }
145. // Cancel is shorthand for CancelAt(time.Now()).
146. func (r *Reservation) Cancel () {
147. r.CancelAt(time.Now())
148. return
149. }
151. // CancelAt indicates that the reservation holder will not perform the reserved action
152. // and reverses the effects of this Reservation on the rate limit as much as possible,
153. // considering that other reservations may have already been made.
154. func (r *Reservation) CancelAt (now time.Time) {
155. if ! r.ok {
156. return
157. }
159. r.lim.mu.Lock()
160. defer r.lim.mu.Unlock()
162. if r.lim.limit == Inf || r.tokens == 0 || r.timeToAct.Before(now ) {
163. return
164. }
166. // calculate tokens to restore
167. // The duration between lim.lastEvent and r.timeToAct tells us how many tokens were reserved
168. // after r was obtained. These tokens should not be restored.
169. restoreTokens := float64(r.tokens) – r.limit.tokensFromDuration (r.lim.lastEvent.Sub(r.timeToAct))
170. if restoreTokens <= 0 {
171. return
172. }
173. // advance time to now
174. now, _, tokens := r.lim.advance(now)
175. // calculate new number of tokens
176. tokens += restoreTokens
177. if burst := float64(r.lim.burst); tokens > burst {
178. tokens = burst
179. }
180. // update state
181. r.lim.last = now
182. r.lim.tokens = tokens
183. if r.timeToAct == r.lim.lastEvent {
184. prevEvent := r.timeToAct.Add(r.limit.durationFromTokens( float64(-r.tokens)))
185. if ! prevEvent.Before(now) {
186. r.lim.lastEvent = prevEvent
187. }
188. }
190. return
191. }
193. // Reserve is shorthand for ReserveN(time.Now(), 1).
194. func (lim *Limiter) Reserve () *Reservation {
195. return lim.ReserveN(time.Now() , 1)
196. }
198. // ReserveN returns a Reservation that indicates how long the caller must wait before n events happen.
199. // The Limiter takes this Reservation into account when allowing future events.
200. // ReserveN returns false if n exceeds the Limiter’s burst size.
201. // Usage example:
202. // r, ok := lim.ReserveN(time.Now(), 1)
203. // if ! ok {
204. // // Not allowed to act! Did you remember to set lim.burst to be > 0 ?
205. // }
206. // time.Sleep(r.Delay())
207. // Act()
208. // Use this method if you wish to wait and slow down in accordance with the rate limit without dropping events.
209. // If you need to respect a deadline or cancel the delay, use Wait instead.
210. // To drop or skip events exceeding rate limit, use Allow instead.
211. func (lim *Limiter) ReserveN (now time.Time, n int) *Reservation {
212. r := lim.reserveN(now, n, InfDuration)
213. return &r
214. }
216. // Wait is shorthand for WaitN(ctx, 1).
217. func (lim *Limiter) Wait (ctx context.Context) (err error) {
218. return lim.WaitN(ctx, 1)
219. }
221. // WaitN blocks until lim permits n events to happen.
222. // It returns an error if n exceeds the Limiter’s burst size, the Context is
223. // canceled, or the expected wait time exceeds the Context’s Deadline.
224. func (lim *Limiter) WaitN (ctx context.Context, n int) (err error) {
225. if n > lim.burst {
226. return fmt.Errorf(“rate: Wait(n=%d) exceeds limiter’s burst %d”, n, lim.burst )
227. }
228. // Check if ctx is already cancelled
229. select {
230. case <-ctx.Done():
231. return ctx.Err()
232. default:
233. }
234. // Determine wait limit
235. now := time.Now()
236. waitLimit := InfDuration
237. if deadline, ok := ctx.Deadline(); ok {
238. waitLimit = deadline.Sub(now)
239. }
240. // Reserve
241. r := lim.reserveN(now, n, waitLimit)
242. if ! r.ok {
243. return fmt.Errorf(“rate: Wait(n=%d) would exceed context deadline”, n )
244. }
245. // Wait
246. t := time.NewTimer(r.DelayFrom(now ))
247. defer t.Stop()
248. select {
249. case <-t.C:
250. // We can proceed.
251. return nil
252. case <-ctx.Done():
253. // Context was canceled before we could proceed. Cancel the
254. // reservation, which may permit other events to proceed sooner.
255. r.Cancel()
256. return ctx.Err()
257. }
258. }
260. // SetLimit is shorthand for SetLimitAt(time.Now(), newLimit).
261. func (lim *Limiter) SetLimit (newLimit Limit) {
262. lim.SetLimitAt(time.Now() , newLimit)
263. }
265. // SetLimitAt sets a new Limit for the limiter. The new Limit, and Burst, may be violated
266. // or underutilized by those which reserved (using Reserve or Wait) but did not yet act
267. // before SetLimitAt was called.
268. func (lim *Limiter) SetLimitAt (now time.Time, newLimit Limit) {
269. lim.mu.Lock()
270. defer lim.mu.Unlock()
272. now, _, tokens := lim.advance(now)
274. lim.last = now
275. lim.tokens = tokens
276. lim.limit = newLimit
277. }
279. // reserveN is a helper method for AllowN, ReserveN, and WaitN.
280. // maxFutureReserve specifies the maximum reservation wait duration allowed.
281. // reserveN returns Reservation, not *Reservation, to avoid allocation in AllowN and WaitN.
282. func (lim *Limiter) reserveN (now time.Time, n int, maxFutureReserve time.Duration) Reservation {
283. lim.mu.Lock()
284. defer lim.mu.Unlock()
286. if lim.limit == Inf {
287. return Reservation{
288. ok: true,
289. lim: lim,
290. tokens: n,
291. timeToAct: now,
292. }
293. }
295. now, last, tokens := lim.advance(now)
297. // Calculate the remaining number of tokens resulting from the request.
298. tokens -= float64(n)
300. // Calculate the wait duration
301. var waitDuration time.Duration
302. if tokens < 0 {
303. waitDuration = lim.limit.durationFromTokens(-tokens)
304. }
306. // Decide result
307. ok := n <= lim.burst && waitDuration <= maxFutureReserve
309. // Prepare reservation
310. r := Reservation{
311. ok: ok,
312. lim: lim,
313. limit: lim.limit,
314. }
315. if ok {
316. r.tokens = n
317. r.timeToAct = now.Add(waitDuration)
318. }
320. // Update state
321. if ok {
322. lim.last = now
323. lim.tokens = tokens
324. lim.lastEvent = r.timeToAct
325. } else {
326. lim.last = last
327. }
329. return r
330. }
332. // advance calculates and returns an updated state for lim resulting from the passage of time.
333. // lim is not changed.
334. func (lim *Limiter) advance(now time.Time) (newNow time.Time, newLast time.Time, newTokens float64 ) {
335. last := lim.last
336. if now.Before(last) {
337. last = now
338. }
340. // Avoid making delta overflow below when last is very old.
341. maxElapsed := lim.limit.durationFromTokens(float64( lim.burst) – lim.tokens)
342. elapsed := now.Sub(last)
343. if elapsed > maxElapsed {
344. elapsed = maxElapsed
345. }
347. // Calculate the new number of tokens, due to time that passed.
348. delta := lim.limit.tokensFromDuration(elapsed)
349. tokens := lim.tokens + delta
350. if burst := float64(lim.burst); tokens > burst {
351. tokens = burst
352. }
354. return now, last, tokens
355. }
357. // durationFromTokens is a unit conversion function from the number of tokens to the duration
358. // of time it takes to accumulate them at a rate of limit tokens per second.
359. func (limit Limit) durationFromTokens(tokens float64) time.Duration {
360. seconds := tokens / float64(limit)
361. return time.Nanosecond * time.Duration(1e9*seconds)
362. }
364. // tokensFromDuration is a unit conversion function from a time duration to the number of tokens
365. // which could be accumulated during that duration at a rate of limit tokens per second.
366. func (limit Limit) tokensFromDuration(d time.Duration) float64 {
367. return d.Seconds() * float64 (limit)
368. }

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.

To play to admire