限流器，顧名思義用來對高并發(fā)的請求進(jìn)行流量限制的組件。

限流包括 Nginx 層面的限流以及業(yè)務(wù)代碼邏輯上的限流。流量的限制在眾多微服務(wù)和 service mesh 中多有應(yīng)用。限流主要有三種算法：信號量、漏桶算法和令牌桶算法。下面依次介紹這三種算法。

筆者在本文的程序示例均以 Go 語言實(shí)現(xiàn)。

1. 問題描述

用戶增長過快、熱門業(yè)務(wù)或者爬蟲等惡意攻擊行為致使請求量突然增大，比如學(xué)校的教務(wù)系統(tǒng)，到了查分之日，請求量漲到之前的 100 倍都不止，沒多久該接口幾乎不可使用，并引發(fā)連鎖反應(yīng)導(dǎo)致整個(gè)系統(tǒng)崩潰。如何應(yīng)對這種情況呢？生活給了我們答案：比如老式電閘都安裝了保險(xiǎn)絲，一旦有人使用超大功率的設(shè)備，保險(xiǎn)絲就會(huì)燒斷以保護(hù)各個(gè)電器不被強(qiáng)電流給燒壞。同理我們的接口也需要安裝上“保險(xiǎn)絲”，以防止非預(yù)期的請求對系統(tǒng)壓力過大而引起的系統(tǒng)癱瘓，當(dāng)流量過大時(shí)，可以采取拒絕或者引流等機(jī)制。

后端服務(wù)由于各個(gè)業(yè)務(wù)的不同和復(fù)雜性，各自在容器部署的時(shí)候都可能會(huì)有單臺(tái)的瓶頸，超過瓶頸會(huì)導(dǎo)致內(nèi)存或者 cpu 的瓶頸，進(jìn)而導(dǎo)致發(fā)生服務(wù)不可用或者單臺(tái)容器直接掛掉或重啟。

2. 信號量限流

信號量在眾多開發(fā)語言中都會(huì)有相關(guān)信號量的設(shè)計(jì)。如 Java 中的Semaphore 是一個(gè)計(jì)數(shù)信號量。常用于限制獲取某資源的線程數(shù)量，可基于 Java 的 concurrent 并發(fā)包實(shí)現(xiàn)。

信號量兩個(gè)重要方法 Acquire() 和 Release()。通過acquire()方法獲取許可，該方法會(huì)阻塞，直到獲取許可為止。通過release()方法釋放許可。

筆者在閱讀一些語言開源實(shí)現(xiàn)后，總結(jié)出信號量的主要有非阻塞和阻塞兩種。

2.1 阻塞方式

采用鎖或者阻塞隊(duì)列方式，以 Go 語言為示例如下：

// 采用channel作為底層數(shù)據(jù)結(jié)構(gòu)，從而達(dá)到阻塞的獲取和使用信號量 type Semaphore struct {innerChan chan struct{} } // 初始化信號量，本質(zhì)初始化一個(gè)channel，channel的初始化大小為 信號量數(shù)值 func NewSemaphore(num uint64) *Semaphore {return &Semaphore{innerChan: make(chan struct{}, num),} } // 獲取信號量，本質(zhì)是 向channel放入元素，如果同時(shí)有很多協(xié)程并發(fā)獲取信號量，則channel則會(huì)full阻塞，從而達(dá)到控制并發(fā)協(xié)程數(shù)的目的，也即是信號量的控制 func (s *Semaphore) Acquire() {for {select {case s.innerChan <- struct{}{}:returndefault:log.Error("semaphore acquire is blocking")time.Sleep(100 * time.Millisecond)}} } // 釋放信號量 本質(zhì)是 從channel中獲取元素，由于有acquire的放入元素，所以此處一定能回去到元素 也就能釋放成功，default只要是出于安全編程的目的 func (s *Semaphore) Release() {select {case <-s.innerChan:returndefault:return} }

在實(shí)現(xiàn)中，定義了 Semaphore 結(jié)構(gòu)體。初始化信號量，本質(zhì)是初始化一個(gè)channel，channel 的初始化大小為信號量數(shù)值；獲取信號量，本質(zhì)是向channel放入元素，如果同時(shí)有很多協(xié)程并發(fā)獲取信號量，則 channel 則會(huì) full 阻塞，從而達(dá)到控制并發(fā)協(xié)程數(shù)的目的，也即是信號量的控制；釋放信號量的本質(zhì)是從channel中獲取元素，由于有acquire的放入元素，所以此處一定能回去到元素 也就能釋放成功，default只要是出于安全編程的目的。

2.2 非阻塞方式

以并發(fā)安全的計(jì)數(shù)方式比如采用原子 atomic 加減進(jìn)行。

3. 限流算法

主流的限流算法分為兩種漏桶算法和令牌桶算法，關(guān)于這兩個(gè)算法有很多文章和論文都給出了詳細(xì)的講解。從原理上看，令牌桶算法和漏桶算法是相反的，一個(gè) 進(jìn)水，一個(gè)是 漏水。值得一提的是 Google Guava 開源和 Uber 開源限流組件均采用漏桶算法。

3.1 漏桶算法

漏桶(Leaky Bucket)算法思路很簡單，水(請求)先進(jìn)入到漏桶里，漏桶以一定的速度出水(接口有響應(yīng)速率)，當(dāng)水流入速度過大會(huì)直接溢出(訪問頻率超過接口響應(yīng)速率)然后就拒絕請求。可以看出漏桶算法能強(qiáng)行限制數(shù)據(jù)的傳輸速率。示意圖如下：

可見這里有兩個(gè)變量，一個(gè)是桶的大小，支持流量突發(fā)增多時(shí)可以存多少的水(burst)，另一個(gè)是水桶漏洞的大小(rate)。

漏桶算法可以使用 redis 隊(duì)列來實(shí)現(xiàn)，生產(chǎn)者發(fā)送消息前先檢查隊(duì)列長度是否超過閾值，超過閾值則丟棄消息，否則發(fā)送消息到 Redis 隊(duì)列中；消費(fèi)者以固定速率從 Redis 隊(duì)列中取消息。Redis 隊(duì)列在這里起到了一個(gè)緩沖池的作用，起到削峰填谷、流量整形的作用。

3.2 令牌桶算法

對于很多應(yīng)用場景來說，除了要求能夠限制數(shù)據(jù)的平均傳輸速率外，還要求允許某種程度的突發(fā)傳輸。這時(shí)候漏桶算法可能就不合適了，令牌桶算法更為適合。令牌桶算法的原理是系統(tǒng)會(huì)以一個(gè)恒定的速度往桶里放入令牌，而如果請求需要被處理，則需要先從桶里獲取一個(gè)令牌，當(dāng)桶里沒有令牌可取時(shí)，則拒絕服務(wù)。桶里能夠存放令牌的最高數(shù)量，就是允許的突發(fā)傳輸量。

放令牌這個(gè)動(dòng)作是持續(xù)不斷的進(jìn)行，如果桶中令牌數(shù)達(dá)到上限，就丟棄令牌，所以就存在這種情況，桶中一直有大量的可用令牌，這時(shí)進(jìn)來的請求就可以直接拿到令牌執(zhí)行，比如設(shè)置qps為100，那么限流器初始化完成一秒后，桶中就已經(jīng)有100個(gè)令牌了，等啟動(dòng)完成對外提供服務(wù)時(shí)，該限流器可以抵擋瞬時(shí)的100個(gè)請求。所以，只有桶中沒有令牌時(shí)，請求才會(huì)進(jìn)行等待，最后相當(dāng)于以一定的速率執(zhí)行。

可以準(zhǔn)備一個(gè)隊(duì)列，用來保存令牌，另外通過一個(gè)線程池定期生成令牌放到隊(duì)列中，每來一個(gè)請求，就從隊(duì)列中獲取一個(gè)令牌，并繼續(xù)執(zhí)行。

3.3 漏桶算法的實(shí)現(xiàn)

所以此處筆者開門見山，直接展示此算法的 Go 語言版本的實(shí)現(xiàn)，代碼如下：

// 此處截取自研的熔斷器代碼中的限流實(shí)現(xiàn)，這是非阻塞的實(shí)現(xiàn) func (sp *servicePanel) incLimit() error {// 如果大于限制的條件則返回錯(cuò)誤if sp.currentLimitCount.Load() > sp.currLimitFunc(nil) {return ErrCurrentLimit}sp.currentLimitCount.Inc()return nil }func (sp *servicePanel) clearLimit() {// 定期每秒重置計(jì)數(shù)器，從而達(dá)到每秒限制的并發(fā)數(shù)// 比如限制1000req/s，在這里指每秒清理1000的計(jì)數(shù)值 // 令牌桶是定期放，這里是逆思維，每秒清空，實(shí)現(xiàn)不僅占用內(nèi)存低而且效率高t := time.NewTicker(time.Second)for {select {case <-t.C:sp.currentLimitCount.Store(0)}} }

上述的實(shí)現(xiàn)實(shí)際是比較粗糙的實(shí)現(xiàn)，沒有嚴(yán)格按照每個(gè)請求方按照某個(gè)固定速率進(jìn)行，而是以秒為單位，粗粒度的進(jìn)行計(jì)數(shù)清零，這其實(shí)會(huì)造成某個(gè)瞬間雙倍的每秒限流個(gè)數(shù)，雖然看上去不滿足要求，但是在這個(gè)瞬間其實(shí)是只是一個(gè)雙倍值，正常系統(tǒng)都應(yīng)該會(huì)應(yīng)付一瞬間雙倍限流個(gè)數(shù)的請求量。

改進(jìn)

如果要嚴(yán)格的按照每個(gè)請求按照某個(gè)固定數(shù)值進(jìn)行，那么可以改進(jìn)時(shí)間的粗力度，具體做法如下：

func (sp *servicePanel) incLimit() error {// 如果大于1則返回錯(cuò)誤if sp.currentLimitCount.Load() > 1 {return ErrCurrentLimit}sp.currentLimitCount.Inc()return nil }func (sp *servicePanel) clearLimit() {// 1s除以每秒限流個(gè)數(shù)t := time.NewTicker(time.Second/time.Duration(sp.currLimitFunc(nil)))for {select {case <-t.C:sp.currentLimitCount.Store(0)}} }

讀者可以自行嘗試一下改進(jìn)之后的漏斗算法。

4. Uber 開源實(shí)現(xiàn) RateLimit 深入解析

uber 在 Github 上開源了一套用于服務(wù)限流的 go 語言庫 ratelimit, 該組件基于 Leaky Bucket(漏桶)實(shí)現(xiàn)。

4.1 引入方式

#第一版本 go get github.com/uber-go/ratelimit@v0.1.0 #改進(jìn)版本 go get github.com/uber-go/ratelimit@master

4.2 使用

首先強(qiáng)調(diào)一點(diǎn)，跟筆者自研的限流器最大的不同的是，這是一個(gè)阻塞調(diào)用者的限流組件。限流速率一般表示為 rate/s 即一秒內(nèi) rate 個(gè)請求。先不多說，進(jìn)行一下用法示例：

func ExampleRatelimit() {rl := ratelimit.New(100) // per secondprev := time.Now()for i := 0; i < 10; i++ {now := rl.Take()if i > 0 {fmt.Println(i, now.Sub(prev))}prev = now} }

預(yù)期的結(jié)果如下：

// Output:// 1 10ms// 2 10ms// 3 10ms// 4 10ms// 5 10ms// 6 10ms// 7 10ms// 8 10ms// 9 10ms

測試結(jié)果完全符合預(yù)期。在這個(gè)例子中，我們給定限流器每秒可以通過100個(gè)請求，也就是平均每個(gè)請求間隔10ms。因此，最終會(huì)每10ms打印一行數(shù)據(jù)。

4.3 實(shí)現(xiàn)細(xì)節(jié)

構(gòu)造限流器

首先是構(gòu)造一個(gè)Limiter 里面有一個(gè) perRequest 這是關(guān)鍵的一個(gè)變量，表示每個(gè)請求之間相差的間隔時(shí)間，這是此組件的算法核心思想，也就是說將請求排隊(duì)，一秒之內(nèi)有rate個(gè)請求，將這些請求排隊(duì)，挨個(gè)來，每個(gè)請求的間隔就是1s/rate 從來達(dá)到 1s內(nèi)rate個(gè)請求的概念，從而達(dá)到限流的目的。

// New returns a Limiter that will limit to the given RPS. func New(rate int, opts ...Option) Limiter {l := &limiter{perRequest: time.Second / time.Duration(rate),maxSlack: -10 * time.Second / time.Duration(rate),}for _, opt := range opts {opt(l)}if l.clock == nil {l.clock = clock.New()}return l }

限流器Take() 阻塞方法

Take() 方法 每次請求前使用，用來獲取批準(zhǔn) 返回批準(zhǔn)時(shí)刻的時(shí)間。

第一版本

// Take blocks to ensure that the time spent between multiple // Take calls is on average time.Second/rate. func (t *limiter) Take() time.Time {t.Lock()defer t.Unlock()now := t.clock.Now()// If this is our first request, then we allow it.if t.last.IsZero() {t.last = nowreturn t.last}// sleepFor calculates how much time we should sleep based on// the perRequest budget and how long the last request took.// Since the request may take longer than the budget, this number// can get negative, and is summed across requests.t.sleepFor += t.perRequest - now.Sub(t.last)// We shouldn't allow sleepFor to get too negative, since it would mean that// a service that slowed down a lot for a short period of time would get// a much higher RPS following that.if t.sleepFor < t.maxSlack {t.sleepFor = t.maxSlack}// If sleepFor is positive, then we should sleep now.if t.sleepFor > 0 {t.clock.Sleep(t.sleepFor)t.last = now.Add(t.sleepFor)t.sleepFor = 0} else {t.last = now}return t.last }

在實(shí)現(xiàn)方面，可以看到第一版本采用了 Go 的 lock，然后排隊(duì) sleep，完成 sleep 之后，請求之間的間隔時(shí)間恒定，單位時(shí)間之內(nèi)有設(shè)定好的請求數(shù)，實(shí)現(xiàn)限流的目的。

第二版本

// Take blocks to ensure that the time spent between multiple // Take calls is on average time.Second/rate. func (t *limiter) Take() time.Time {newState := state{}taken := falsefor !taken {now := t.clock.Now()previousStatePointer := atomic.LoadPointer(&t.state)oldState := (*state)(previousStatePointer)newState = state{}newState.last = now// If this is our first request, then we allow it.if oldState.last.IsZero() {taken = atomic.CompareAndSwapPointer(&t.state, previousStatePointer, unsafe.Pointer(&newState))continue}// sleepFor calculates how much time we should sleep based on// the perRequest budget and how long the last request took.// Since the request may take longer than the budget, this number// can get negative, and is summed across requests.newState.sleepFor += t.perRequest - now.Sub(oldState.last)// We shouldn't allow sleepFor to get too negative, since it would mean that// a service that slowed down a lot for a short period of time would get// a much higher RPS following that.if newState.sleepFor < t.maxSlack {newState.sleepFor = t.maxSlack}if newState.sleepFor > 0 {newState.last = newState.last.Add(newState.sleepFor)}taken = atomic.CompareAndSwapPointer(&t.state, previousStatePointer, unsafe.Pointer(&newState))}t.clock.Sleep(newState.sleepFor)return newState.last }

第二版本采用原子操作+for的自旋操作來替代lock操作，這樣做的目的是減少協(xié)程鎖競爭。 兩個(gè)版本不管是用鎖還是原子操作本質(zhì)都是讓請求排隊(duì)，第一版本存在鎖競爭，然后排隊(duì)sleep，第二版本避免鎖競爭，但是所有協(xié)程可能很快跳出for循環(huán)然后都會(huì)在sleep處sleep。

5. 小結(jié)

保障服務(wù)穩(wěn)定的三大利器：熔斷降級、服務(wù)限流和故障模擬。本文主要講解了分布式系統(tǒng)中高可用的常用策略：限流。限流通常有三種實(shí)現(xiàn)：信號量（計(jì)數(shù)器）、漏桶、令牌桶。本文基于漏桶算法實(shí)現(xiàn)了一個(gè)限流小插件。最后分析了 uber 開源的 uber-go，限流器 Take() 阻塞方法的第二版本對協(xié)程鎖競爭更加友好。

鏈接：https://juejin.im/post/5eb2cfcce51d4528dd23bd7e

總結(jié)

以上是生活随笔為你收集整理的jq实现文字个数限制_分布式系统高可用实战之限流器（Go 版本实现）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。