項目中用到了限流，受限于一些實現方式上的東西，手撕了一個簡單的服務端限流器。
服務端限流和客戶端限流的區別，簡單來說就是：
1）服務端限流
對接口請求進行限流，限制的是單位時間內請求的數量，目的是通過有損來換取高可用。
例如我們的場景是，有一個服務接收請求，處理之后，將數據bulk到Elasticsearch中進行索引存儲，bulk索引是一個很耗費資源的操作，如果遭遇到請求流量激增，可能會壓垮Elasticsearch（隊列阻塞，內存激增），所以需要對流量的峰值做一個限制。
2）客戶端限流
限制的是客戶端進行訪問的次數。
例如，線程池就是一個天然的限流器。限制了并發個數max_connection，多了的就放到緩沖隊列里排隊，排隊擱不下了>queue_size就扔掉。
本文是服務端限流器。
我這個限流器的優點：
1）簡單
2）管事
缺點：
1）不能做到平滑限流
例如大家嘗嘗說的令牌桶算法和漏桶算法（我感覺這兩個算法本質上都是一個事情）可以實現平滑限流。什么是平滑限流？舉個栗子，我們要限制5秒鐘內訪問數不超過1000，平滑限流能做到，每秒200個，5秒鐘不超過1000，很平衡；非平滑限流可能，在第一秒就訪問了1000次，之后的4秒鐘全部限制住。•2）不靈活
只實現了秒級的限流。
支持兩個場景：
1）對于單進程多線程場景（使用線程安全的Queue做全局變量）
這種場景下，只部署了一個實例，對這個實例進行限流。在生產環境中用的很少。
2）對于多進程分布式場景（使用redis做全局變量）
多實例部署，一般來說生產環境，都是這樣的使用場景。
在這樣的場景下，需要對流量進行整體的把控。例如，user服務部署了三個實例，對外暴露query接口，要做的是對接口級的流量限制，也就是對query這個接口整體允許多大的峰值，而不去關心到底負載到哪個實例。
題外話，這個可以通過nginx做。
下面說一下限流器的實現吧。
1、接口BaseRateLimiter
按照我的思路，先定義一個接口，也可以叫抽象類。
初始化的時候，要配置rate，限流器的限速。
提供一個抽象方法，acquire()，調用這個方法，返回是否限制流量。
|
1
2
3
4
5
6
7
8
9
10
11
|
class BaseRateLimiter(object): __metaclass__ = abc.ABCMeta @abc.abstractmethod def __init__(self, rate): self.rate = rate @abc.abstractmethod def acquire(self, count): return |
2、單進程多線程場景的限流ThreadingRateLimiter
繼承BaseRateLimiter抽象類，使用線程安全的Queue作為全局變量，來消除競態影響。
后臺有個進程每秒鐘清空一次queue；
當請求來了，調用acquire函數，queue incr一次，如果大于限速了，就返回限制。否則就允許訪問。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
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、分布式場景下的限流DistributeRateLimiter
繼承BaseRateLimiter抽象類，使用外部存儲作為共享變量，外部存儲的訪問方式為cache。
|
1
2
3
4
5
6
7
8
9
10
11
12
|
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 |
為了解耦和靈活性，我們實現了Cache類。提供一個抽象方法getToken()
如果你使用redis的話，你就繼承Cache抽象類，實現通過redis獲取令牌的方法。
如果使用mysql的話，你就繼承Cache抽象類，實現通過mysql獲取令牌的方法。
cache抽象類
|
1
2
3
4
5
6
7
8
9
10
11
12
|
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 |
給出一個redis的實現RedisTokenCache
每秒鐘創建一個key，并且對請求進行計數incr，當這一秒的計數值已經超過了限速rate，就拿不到token了，也就是限制流量。
對每秒鐘創建出的key，讓他超時expire。保證key不會持續占用存儲空間。
沒有什么難點，這里使用redis事務，保證incr和expire能同時執行成功。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
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 |
多線程場景下測試代碼
|
1
2
3
4
5
6
7
8
9
10
11
12
|
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() |
分布式場景下測試代碼
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
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() |
可以自行跑一下。
說明：
我這里的限速都是秒級別的，例如限制每秒400次請求。有可能出現這一秒的前100ms，就來了400次請求，后900ms就全部限制住了。也就是不能平滑限流。
不過如果你后臺的邏輯有隊列，或者線程池這樣的緩沖，這個不平滑的影響其實不大。
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持服務器之家。
原文鏈接：http://www.cnblogs.com/kangoroo/p/7700758.html?utm_source=tuicool&utm_medium=referral
