2. 程式人生 > 實用技巧 >Golang實現請求限流的幾種辦法

Golang實現請求限流的幾種辦法

阿新 • • 發佈：2020-11-01

目錄

簡單的併發控制

使用計數器實現請求限流

使用golang官方包實現httpserver頻率限制

使用Token Bucket（令牌桶演算法）實現請求限流

簡單的併發控制

利用 channel 的緩衝設定，我們就可以來實現併發的限制。我們只要在執行併發的同時，往一個帶有緩衝的channel裡寫入點東西（隨便寫啥，內容不重要）。讓併發的goroutine在執行完成後把這個channel裡的東西給讀走。這樣整個併發的數量就講控制在這個channel的緩衝區大小上。

比如我們可以用一個bool型別的帶緩衝channel作為併發限制的計數器。

chLimit := make(chan bool, 1)

然後在併發執行的地方，每建立一個新的 goroutine，都往chLimit裡塞個東西。

  1. for i, sleeptime := range input {
  2. chs[i] = make(chan string, 1)
  3. chLimit <- true
  4. go limitFunc(chLimit, chs[i], i, sleeptime, timeout)
  5. }

這裡通過go關鍵字併發執行的是新構造的函式。他在執行完後，會把chLimit的緩衝區裡給消費掉一個。

  1. limitFunc := func(chLimit chan bool, ch chan string, task_id, sleeptime, timeout int) {
  2. Run(task_id, sleeptime, timeout, ch)
  3. <-chLimit
  4. }

這樣一來，當建立的goroutine數量到達chLimit的緩衝區上限後。主goroutine就掛起阻塞了，直到這些goroutine執行完畢，消費掉了chLimit緩衝區中的資料，程式才會繼續建立新的goroutine。我們併發數量限制的目的也就達到了。

以下是完整程式碼：

  1. package main
  3. import (
  4. "fmt"
  5. "time"
  6. )
  8. func Run(task_id, sleeptime, timeout int, ch chan string) {
  9. ch_run := make(chan string)
  10. go run(task_id, sleeptime, ch_run)
  11. select {
  12. case re := <-ch_run:
  13. ch <- re
  14. case <-time.After(time.Duration(timeout) * time.Second):
  15. re := fmt.Sprintf("task id %d , timeout", task_id)
  16. ch <- re
  17. }
  18. }
  20. func run(task_id, sleeptime int, ch chan string) {
  22. time.Sleep(time.Duration(sleeptime) * time.Second)
  23. ch <- fmt.Sprintf("task id %d , sleep %d second", task_id, sleeptime)
  24. return
  25. }
  27. func main() {
  28. input := []int{3, 2, 1}
  29. timeout := 2
  30. chLimit := make(chan bool, 1)
  31. chs := make([]chan string, len(input))
  32. limitFunc := func(chLimit chan bool, ch chan string, task_id, sleeptime, timeout int) {
  33. Run(task_id, sleeptime, timeout, ch)
  34. <-chLimit
  35. }
  36. startTime := time.Now()
  37. fmt.Println("Multirun start")
  38. for i, sleeptime := range input {
  39. chs[i] = make(chan string, 1)
  40. chLimit <- true
  41. go limitFunc(chLimit, chs[i], i, sleeptime, timeout)
  42. }
  44. for _, ch := range chs {
  45. fmt.Println(<-ch)
  46. }
  47. endTime := time.Now()
  48. fmt.Printf("Multissh finished. Process time %s. Number of task is %d", endTime.Sub(startTime), len(input))
  49. }

執行結果：

  1. Multirun start
  2. task id 0 , timeout
  3. task id 1 , timeout
  4. task id 2 , sleep 1 second
  5. Multissh finished. Process time 5s. Number of task is 3

如果修改併發限制為2：

chLimit := make(chan bool, 2)

執行結果：

  2. Multirun start
  3. task id 0 , timeout
  4. task id 1 , timeout
  5. task id 2 , sleep 1 second
  6. Multissh finished. Process time 3s. Number of task is 3

使用計數器實現請求限流

限流的要求是在指定的時間間隔內，server 最多隻能服務指定數量的請求。實現的原理是我們啟動一個計數器，每次服務請求會把計數器加一，同時到達指定的時間間隔後會把計數器清零；這個計數器的實現程式碼如下所示：

  1. type RequestLimitService struct {
  2. Interval time.Duration
  3. MaxCount int
  4. Lock sync.Mutex
  5. ReqCount int
  6. }
  8. func NewRequestLimitService(interval time.Duration, maxCnt int) *RequestLimitService {
  9. reqLimit := &RequestLimitService{
  10. Interval: interval,
  11. MaxCount: maxCnt,
  12. }
  14. go func() {
  15. ticker := time.NewTicker(interval)
  16. for {
  17. <-ticker.C
  18. reqLimit.Lock.Lock()
  19. fmt.Println("Reset Count...")
  20. reqLimit.ReqCount = 0
  21. reqLimit.Lock.Unlock()
  22. }
  23. }()
  25. return reqLimit
  26. }
  28. func (reqLimit *RequestLimitService) Increase() {
  29. reqLimit.Lock.Lock()
  30. defer reqLimit.Lock.Unlock()
  32. reqLimit.ReqCount += 1
  33. }
  35. func (reqLimit *RequestLimitService) IsAvailable() bool {
  36. reqLimit.Lock.Lock()
  37. defer reqLimit.Lock.Unlock()
  39. return reqLimit.ReqCount < reqLimit.MaxCount
  40. }

在服務請求的時候, 我們會對當前計數器和閾值進行比較，只有未超過閾值時才進行服務：

  1. var RequestLimit = NewRequestLimitService(10 * time.Second, 5)
  3. func helloHandler(w http.ResponseWriter, r *http.Request) {
  4. if RequestLimit.IsAvailable() {
  5. RequestLimit.Increase()
  6. fmt.Println(RequestLimit.ReqCount)
  7. io.WriteString(w, "Hello world!\n")
  8. } else {
  9. fmt.Println("Reach request limiting!")
  10. io.WriteString(w, "Reach request limit!\n")
  11. }
  12. }
  14. func main() {
  15. fmt.Println("Server Started!")
  16. http.HandleFunc("/", helloHandler)
  17. http.ListenAndServe(":8000", nil)
  18. }

完整程式碼url：https://github.com/hiberabyss/JustDoIt/blob/master/RequestLimit/request_limit.go

使用golang官方包實現httpserver頻率限制

使用golang來編寫httpserver時，可以使用官方已經有實現好的包：

  1. import(
  2. "fmt"
  3. "net"
  4. "golang.org/x/net/netutil"
  5. )
  7. func main() {
  8. l, err := net.Listen("tcp", "127.0.0.1:0")
  9. if err != nil {
  10. fmt.Fatalf("Listen: %v", err)
  11. }
  12. defer l.Close()
  13. l = LimitListener(l, max)
  15. http.Serve(l, http.HandlerFunc())
  17. //bla bla bla.................
  18. }

原始碼如下(url :https://github.com/golang/net/blob/master/netutil/listen.go)，基本思路就是為連線數計數，通過make chan來建立一個最大連線數的channel, 每次accept就+1，close時候就-1. 當到達最大連線數時，就等待空閒連接出來之後再accept。

  1. // Copyright 2013 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 netutil provides network utility functions, complementing the more
  6. // common ones in the net package.
  7. package netutil // import "golang.org/x/net/netutil"
  9. import (
  10. "net"
  11. "sync"
  12. )
  14. // LimitListener returns a Listener that accepts at most n simultaneous
  15. // connections from the provided Listener.
  16. func LimitListener(l net.Listener, n int) net.Listener {
  17. return &limitListener{
  18. Listener: l,
  19. sem: make(chan struct{}, n),
  20. done: make(chan struct{}),
  21. }
  22. }
  24. type limitListener struct {
  25. net.Listener
  26. sem chan struct{}
  27. closeOnce sync.Once // ensures the done chan is only closed once
  28. done chan struct{} // no values sent; closed when Close is called
  29. }
  31. // acquire acquires the limiting semaphore. Returns true if successfully
  32. // accquired, false if the listener is closed and the semaphore is not
  33. // acquired.
  34. func (l *limitListener) acquire() bool {
  35. select {
  36. case <-l.done:
  37. return false
  38. case l.sem <- struct{}{}:
  39. return true
  40. }
  41. }
  42. func (l *limitListener) release() { <-l.sem }
  44. func (l *limitListener) Accept() (net.Conn, error) {
  45. //如果sem滿了，就會阻塞在這
  46. acquired := l.acquire()
  47. // If the semaphore isn't acquired because the listener was closed, expect
  48. // that this call to accept won't block, but immediately return an error.
  49. c, err := l.Listener.Accept()
  50. if err != nil {
  51. if acquired {
  52. l.release()
  53. }
  54. return nil, err
  55. }
  56. return &limitListenerConn{Conn: c, release: l.release}, nil
  57. }
  59. func (l *limitListener) Close() error {
  60. err := l.Listener.Close()
  61. l.closeOnce.Do(func() { close(l.done) })
  62. return err
  63. }
  65. type limitListenerConn struct {
  66. net.Conn
  67. releaseOnce sync.Once
  68. release func()
  69. }
  71. func (l *limitListenerConn) Close() error {
  72. err := l.Conn.Close()
  73. //close時釋放佔用的sem
  74. l.releaseOnce.Do(l.release)
  75. return err
  76. }

使用Token Bucket（令牌桶演算法）實現請求限流

在開發高併發系統時有三把利器用來保護系統：快取、降級和限流!為了保證在業務高峰期，線上系統也能保證一定的彈性和穩定性，最有效的方案就是進行服務降級了，而限流就是降級系統最常採用的方案之一。

這裡為大家推薦一個開源庫https://github.com/didip/tollbooth，但是，如果您想要一些簡單的、輕量級的或者只是想要學習的東西，實現自己的中介軟體來處理速率限制並不困難。今天我們就來聊聊如何實現自己的一個限流中介軟體

首先我們需要安裝一個提供了Token bucket(令牌桶演算法)的依賴包，上面提到的toolbooth 的實現也是基於它實現的：

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

先看Demo程式碼的實現：

  1. package main
  3. import (
  4. "net/http"
  5. "golang.org/x/time/rate"
  6. )
  8. var limiter = rate.NewLimiter(2, 5)
  9. func limit(next http.Handler) http.Handler {
  10. return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
  11. if limiter.Allow() == false {
  12. http.Error(w, http.StatusText(429), http.StatusTooManyRequests)
  13. return
  14. }
  15. next.ServeHTTP(w, r)
  16. })
  17. }
  19. func main() {
  20. mux := http.NewServeMux()
  21. mux.HandleFunc("/", okHandler)
  22. // Wrap the servemux with the limit middleware.
  23. http.ListenAndServe(":4000", limit(mux))
  24. }
  26. func okHandler(w http.ResponseWriter, r *http.Request) {
  27. w.Write([]byte("OK"))
  28. }

然後看看rate.NewLimiter的原始碼：

演算法描述：使用者配置的平均傳送速率為r，則每隔1/r秒一個令牌被加入到桶中（每秒會有r個令牌放入桶中），桶中最多可以存放b個令牌。如果令牌到達時令牌桶已經滿了，那麼這個令牌會被丟棄；

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

雖然在某些情況下使用單個全域性速率限制器非常有用，但另一種常見情況是基於IP地址或API金鑰等識別符號為每個使用者實施速率限制器。我們將使用IP地址作為識別符號。簡單實現程式碼如下：

  1. package main
  2. import (
  3. "net/http"
  4. "sync"
  5. "time"
  6. "golang.org/x/time/rate"
  7. )
  8. // Create a custom visitor struct which holds the rate limiter for each
  9. // visitor and the last time that the visitor was seen.
  10. type visitor struct {
  11. limiter *rate.Limiter
  12. lastSeen time.Time
  13. }
  14. // Change the the map to hold values of the type visitor.
  15. var visitors = make(map[string]*visitor)
  16. var mtx sync.Mutex
  17. // Run a background goroutine to remove old entries from the visitors map.
  18. func init() {
  19. go cleanupVisitors()
  20. }
  21. func addVisitor(ip string) *rate.Limiter {
  22. limiter := rate.NewLimiter(2, 5)
  23. mtx.Lock()
  24. // Include the current time when creating a new visitor.
  25. visitors[ip] = &visitor{limiter, time.Now()}
  26. mtx.Unlock()
  27. return limiter
  28. }
  29. func getVisitor(ip string) *rate.Limiter {
  30. mtx.Lock()
  31. v, exists := visitors[ip]
  32. if !exists {
  33. mtx.Unlock()
  34. return addVisitor(ip)
  35. }
  36. // Update the last seen time for the visitor.
  37. v.lastSeen = time.Now()
  38. mtx.Unlock()
  39. return v.limiter
  40. }
  41. // Every minute check the map for visitors that haven't been seen for
  42. // more than 3 minutes and delete the entries.
  43. func cleanupVisitors() {
  44. for {
  45. time.Sleep(time.Minute)
  46. mtx.Lock()
  47. for ip, v := range visitors {
  48. if time.Now().Sub(v.lastSeen) > 3*time.Minute {
  49. delete(visitors, ip)
  50. }
  51. }
  52. mtx.Unlock()
  53. }
  54. }
  55. func limit(next http.Handler) http.Handler {
  56. return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
  57. limiter := getVisitor(r.RemoteAddr)
  58. if limiter.Allow() == false {
  59. http.Error(w, http.StatusText(429), http.StatusTooManyRequests)
  60. return
  61. }
  62. next.ServeHTTP(w, r)
  63. })
  64. }

轉載:https://blog.csdn.net/micl200110041/article/details/82013032

相關推薦

Golang實現請求限流的幾種辦法(小結)

在開發高併發系統時，有三把利器用來保護系統：快取、降級和限流。那麼何為限流呢？顧名思義，限流就是限制流量，就像你寬頻包了1個G的流量，用完了就沒了。

Golang實現請求限流的幾種辦法

目錄 簡單的併發控制 使用計數器實現請求限流 使用golang官方包實現httpserver頻率限制

3種方式實現Redis限流

Redis限流的實現方式有3種，分別是：1、基於Redis的setnx的操作，給指定的key設定了過期實踐；2、基於Redis的資料結構zset，將請求打造成一個zset陣列；3、基於Redis的令牌桶演算法，輸出速率大於輸入速率，就要限流

Redis和Lua實現分散式限流器的方法詳解

主要是依靠 redis + lua 來實現限流器,使用 lua 的原因是將多條命令合併在一起作為一個原子操作,無需過多考慮併發.

Python要如何實現列表排序的幾種方法

排序，是許多程式語言中經常出現的問題。同樣的，在Python中，如何是實現排序呢？（以下排序都是基於列表來實現）

Golang實現超時退出的三種方式

前段時間發現線上有個服務介面，總是間歇性告警，有時候一天兩三次，有時候一天都沒有。

Spring Cloud Alibaba系列（五）sentinel實現服務限流降級

一、sentinel是什麼 sentinel的官方名稱叫分散式系統的流量防衛兵。Sentinel 以流量為切入點，從流量控制、熔斷降級、系統負載保護等多個維度保護服務的穩定性。在Spring Cloud專案中最開始我們使用的是Hystrix，目前

ElementUI 修改預設樣式的幾種辦法(小結)

ElementUI 是一套ui元件庫，目前最新版本 react 和 vue 等主流框架都有支援。該庫預設主題色是天藍色，若用於專案開發，難免遇到要需求修改其預設樣式的情況，本文就基於 react 和 vue 框架介紹幾種修改 ElementUI 預

web頁面實現檔案下載的幾種方式

參考：https://www.jianshu.com/p/bf0a4e3926a4https://blog.csdn.net/qq_30944053/article/details/98761447

js實現深拷貝的幾種方法

實現陣列深拷貝的一些方法： 1.使用es6 var a=[1,2,3] var b=[...a]; b.push(4); console.log(b);//1,2,3,4

後端請求介面的幾種方式

//方法A：HttpClient（首選） using System.Net.Http; private static readonly HttpClient client = new HttpClient();

實現深拷貝的幾種方式

目前使用過四種 1、遞迴遞迴去複製所有層級屬性 function deepClone(obj){ let objClone = Array.isArray(obj)?[]:{};

SpringBoot服務上實現介面限流的方法

Sentinel是阿里巴巴開源的限流器熔斷器，並且帶有視覺化操作介面。 在日常開發中，限流功能時常被使用，用於對某些介面進行限流熔斷，譬如限制單位時間內介面訪問次數；或者按照某種規則進行限流，如限制ip的單位時間

判斷是否是陣列的幾種辦法

1.通過instanceof判斷 instanceof運算子用於檢驗建構函式的prototype屬性是否出現在物件的原型鏈中的任何位置，返回一個布林值。

SpringBoot基於Sentinel在服務上實現介面限流

Sentinel是阿里巴巴開源的限流器熔斷器，並且帶有視覺化操作介面。 在日常開發中，限流功能時常被使用，用於對某些介面進行限流熔斷，譬如限制單位時間內介面訪問次數；或者按照某種規則進行限流，如限制ip的單位時間

基於Redis+Lua指令碼實現分散式限流元件封裝的方法

建立限流元件專案 pom.xml檔案中引入相關依賴 <dependencies> <dependency> <groupId>org.springframework.boot</groupId>

LBS——實現附近功能的幾種方案淺談

一、背景 附近的人、景點，也即POI，這種需求多見於LBS中，這塊不是很專業，也沒有特別去實現過，只是根據網路查詢的相關網友的說法，將幾種實現手段羅列一下。 前提條件說明，做這種需要興趣點的經緯度資訊，以及需

詳解Springboot整合sentinel實現介面限流入門

Sentinel是阿里巴巴開源的限流器熔斷器，並且帶有視覺化操作介面。 在日常開發中，限流功能時常被使用，用於對某些介面進行限流熔斷，譬如限制單位時間內介面訪問次數；或者按照某種規則進行限流，如限制ip的單位時間

python爬蟲多次請求超時的幾種重試方法(6種)

第一種方法 headers = Dict() url = \'https://www.baidu.com\' try: proxies = None response = requests.get(url,headers=headers,verify=False,proxies=None,timeout=3)

利用 Django 動態展示 Pyecharts 快速實現圖表資料的幾種方法

本文的文字及圖片來源於網路,僅供學習、交流使用,不具有任何商業用途,版權歸原作者所有,如有問題請及時聯絡我們以作處理