項(xiàng)目中用到了限流，受限于一些實(shí)現(xiàn)方式上的東西，手撕了一個(gè)簡(jiǎn)單的服務(wù)端限流器。

服務(wù)端限流和客戶(hù)端限流的區(qū)別，簡(jiǎn)單來(lái)說(shuō)就是：

1）服務(wù)端限流

對(duì)接口請(qǐng)求進(jìn)行限流，限制的是單位時(shí)間內(nèi)請(qǐng)求的數(shù)量，目的是通過(guò)有損來(lái)?yè)Q取高可用。

例如我們的場(chǎng)景是，有一個(gè)服務(wù)接收請(qǐng)求，處理之后，將數(shù)據(jù)bulk到Elasticsearch中進(jìn)行索引存儲(chǔ)，bulk索引是一個(gè)很耗費(fèi)資源的操作，如果遭遇到請(qǐng)求流量激增，可能會(huì)壓垮Elasticsearch（隊(duì)列阻塞，內(nèi)存激增），所以需要對(duì)流量的峰值做一個(gè)限制。

2）客戶(hù)端限流

限制的是客戶(hù)端進(jìn)行訪問(wèn)的次數(shù)。

例如，線程池就是一個(gè)天然的限流器。限制了并發(fā)個(gè)數(shù)max_connection，多了的就放到緩沖隊(duì)列里排隊(duì)，排隊(duì)擱不下了>queue_size就扔掉。

本文是服務(wù)端限流器。

我這個(gè)限流器的優(yōu)點(diǎn)：

1）簡(jiǎn)單
2）管事

缺點(diǎn)：

1）不能做到平滑限流

例如大家嘗嘗說(shuō)的令牌桶算法和漏桶算法（我感覺(jué)這兩個(gè)算法本質(zhì)上都是一個(gè)事情）可以實(shí)現(xiàn)平滑限流。什么是平滑限流？舉個(gè)栗子，我們要限制5秒鐘內(nèi)訪問(wèn)數(shù)不超過(guò)1000，平滑限流能做到，每秒200個(gè)，5秒鐘不超過(guò)1000，很平衡；非平滑限流可能，在第一秒就訪問(wèn)了1000次，之后的4秒鐘全部限制住。•2）不靈活

只實(shí)現(xiàn)了秒級(jí)的限流。

支持兩個(gè)場(chǎng)景：

1）對(duì)于單進(jìn)程多線程場(chǎng)景（使用線程安全的Queue做全局變量）

這種場(chǎng)景下，只部署了一個(gè)實(shí)例，對(duì)這個(gè)實(shí)例進(jìn)行限流。在生產(chǎn)環(huán)境中用的很少。

2）對(duì)于多進(jìn)程分布式場(chǎng)景（使用redis做全局變量）

多實(shí)例部署，一般來(lái)說(shuō)生產(chǎn)環(huán)境，都是這樣的使用場(chǎng)景。

在這樣的場(chǎng)景下，需要對(duì)流量進(jìn)行整體的把控。例如，user服務(wù)部署了三個(gè)實(shí)例，對(duì)外暴露query接口，要做的是對(duì)接口級(jí)的流量限制，也就是對(duì)query這個(gè)接口整體允許多大的峰值，而不去關(guān)心到底負(fù)載到哪個(gè)實(shí)例。

題外話，這個(gè)可以通過(guò)nginx做。 

下面說(shuō)一下限流器的實(shí)現(xiàn)吧。 

1、接口BaseRateLimiter

按照我的思路，先定義一個(gè)接口，也可以叫抽象類(lèi)。

初始化的時(shí)候，要配置rate，限流器的限速。

提供一個(gè)抽象方法，acquire()，調(diào)用這個(gè)方法，返回是否限制流量。

class BaseRateLimiter(object):

  __metaclass__ = abc.ABCMeta

  @abc.abstractmethod
  def __init__(self, rate):
    self.rate = rate

  @abc.abstractmethod
  def acquire(self, count):
    return

2、單進(jìn)程多線程場(chǎng)景的限流ThreadingRateLimiter

繼承BaseRateLimiter抽象類(lèi)，使用線程安全的Queue作為全局變量，來(lái)消除競(jìng)態(tài)影響。

后臺(tái)有個(gè)進(jìn)程每秒鐘清空一次queue；

當(dāng)請(qǐng)求來(lái)了，調(diào)用acquire函數(shù)，queue incr一次，如果大于限速了，就返回限制。否則就允許訪問(wèn)。

class ThreadingRateLimiter(BaseRateLimiter):

  def __init__(self, rate):
    BaseRateLimiter.__init__(self, rate)
    self.queue = Queue.Queue()
    threading.Thread(target=self._clear_queue).start()

  def acquire(self, count=1):
    self.queue.put(1, block=False)
    return self.queue.qsize() < self.rate

  def _clear_queue(self):
    while 1:
      time.sleep(1)
      self.queue.queue.clear()

2、分布式場(chǎng)景下的限流DistributeRateLimiter

繼承BaseRateLimiter抽象類(lèi)，使用外部存儲(chǔ)作為共享變量，外部存儲(chǔ)的訪問(wèn)方式為cache。

class DistributeRateLimiter(BaseRateLimiter):

  def __init__(self, rate, cache):
    BaseRateLimiter.__init__(self, rate)
    self.cache = cache

  def acquire(self, count=1, expire=3, key=None, callback=None):
    try:
      if isinstance(self.cache, Cache):
        return self.cache.fetchToken(rate=self.rate, count=count, expire=expire, key=key)
    except Exception, ex:
      return True

為了解耦和靈活性，我們實(shí)現(xiàn)了Cache類(lèi)。提供一個(gè)抽象方法getToken()

如果你使用redis的話，你就繼承Cache抽象類(lèi)，實(shí)現(xiàn)通過(guò)redis獲取令牌的方法。

如果使用mysql的話，你就繼承Cache抽象類(lèi)，實(shí)現(xiàn)通過(guò)mysql獲取令牌的方法。

cache抽象類(lèi)

class Cache(object):

  __metaclass__ = abc.ABCMeta

  @abc.abstractmethod
  def __init__(self):
    self.key = "DEFAULT"
    self.namespace = "RATELIMITER"

  @abc.abstractmethod
  def fetchToken(self, rate, key=None):
    return

給出一個(gè)redis的實(shí)現(xiàn)RedisTokenCache

每秒鐘創(chuàng)建一個(gè)key，并且對(duì)請(qǐng)求進(jìn)行計(jì)數(shù)incr，當(dāng)這一秒的計(jì)數(shù)值已經(jīng)超過(guò)了限速rate，就拿不到token了，也就是限制流量。

對(duì)每秒鐘創(chuàng)建出的key，讓他超時(shí)expire。保證key不會(huì)持續(xù)占用存儲(chǔ)空間。

沒(méi)有什么難點(diǎn)，這里使用redis事務(wù)，保證incr和expire能同時(shí)執(zhí)行成功。

class RedisTokenCache(Cache):

  def __init__(self, host, port, db=0, password=None, max_connections=None):
    Cache.__init__(self)
    self.redis = redis.Redis(
      connection_pool=
        redis.ConnectionPool(
          host=host, port=port, db=db,
          password=password,
          max_connections=max_connections
        ))

  def fetchToken(self, rate=100, count=1, expire=3, key=None):
    date = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    key = ":".join([self.namespace, key if key else self.key, date])
    try:
      current = self.redis.get(key)
      if int(current if current else "0") > rate:
        raise Exception("to many requests in current second: %s" % date)
      else:
        with self.redis.pipeline() as p:
          p.multi()
          p.incr(key, count)
          p.expire(key, int(expire if expire else "3"))
          p.execute()
          return True
    except Exception, ex:
      return False

多線程場(chǎng)景下測(cè)試代碼 

limiter = ThreadingRateLimiter(rate=10000)

def job():
  while 1:
    if not limiter.acquire():
      print '限流'
    else:
      print '正常'

threads = [threading.Thread(target=job) for i in range(10)]
for thread in threads:
  thread.start()

分布式場(chǎng)景下測(cè)試代碼

token_cache = RedisTokenCache(host='10.93.84.53', port=6379, password='bigdata123')
limiter = DistributeRateLimiter(rate=10000, cache=token_cache)
r = redis.Redis(connection_pool=redis.ConnectionPool(host='10.93.84.53', port=6379, password='bigdata123'))

def job():
  while 1:
    if not limiter.acquire():
      print '限流'
    else:
      print '正常'

threads = [multiprocessing.Process(target=job) for i in range(10)]
for thread in threads:
  thread.start()

可以自行跑一下。 

說(shuō)明：

我這里的限速都是秒級(jí)別的，例如限制每秒400次請(qǐng)求。有可能出現(xiàn)這一秒的前100ms，就來(lái)了400次請(qǐng)求，后900ms就全部限制住了。也就是不能平滑限流。

不過(guò)如果你后臺(tái)的邏輯有隊(duì)列，或者線程池這樣的緩沖，這個(gè)不平滑的影響其實(shí)不大。

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

最新評(píng)論