基於Memcache的限速算法？ （令牌桶？）

Question

我正在尋找一種有效的方法來限制從Google App Engine向第三方服務的請求。第三方服務費率按每個帳戶限制請求，並且在Google App Engine方面，大部分工作都是在任務內執行的。令牌桶在這裏是一個很好的通用算法。

問：有什麼方法可以有效地對每個帳戶的請求進行速率限制，而不是每個服務的限制？

這不應該涉及在GAE任務隊列上設置費率，因爲每個賬戶的請求數量和服務賬戶數量會有很大差異。出於性能原因，我最喜歡基於memcache（incr/decr？）的想法！

我認爲這歸結爲基於memcache的令牌桶？

想法？

Answer 1

我一直在這個項目作爲一個書籤前一陣子：http://code.google.com/p/gaedjango-ratelimitcache/

不是一個真正的回答你的具體問題，但也許它可以幫助您開始。

Answer 2

我知道這是一個老問題，但它是一個頂級的搜索結果，我認爲其他人可能會找到一個替代方案，使我變得有用。這比上面的解決方案有點更精細（精確到秒），簡單（只有一個單一的功能），以及高性能的（只有一個內存緩存查找）：

import webapp2 
from functools import wraps 
from google.appengine.api import memcache 


def rate_limit(seconds_per_request=1): 
    def rate_limiter(function): 
    @wraps(function) 
    def wrapper(self, *args, **kwargs): 
     added = memcache.add('%s:%s' % (self.__class__.__name__, self.request.remote_addr or ''), 1, 
          time=seconds_per_request, namespace='rate_limiting') 
     if not added: 
     self.response.write('Rate limit exceeded.') 
     self.response.set_status(429) 
     return 
     return function(self, *args, **kwargs) 
    return wrapper 
    return rate_limiter 


class ExampleHandler(webapp2.RequestHandler): 
    @rate_limit(seconds_per_request=2) 
    def get(self): 
    self.response.write('Hello, webapp2!') 

Answer 3

這裏是我是如何實現令牌桶的memcache上GAE：

編輯：以（另一個）刺傷此。

這部分地借用https://github.com/simonw/ratelimitcache/blob/master/ratelimitcache.py

def throttle(key, rate_count, rate_seconds, tries=3): 
    ''' 
    returns True if throttled (not enough tokens available) else False 
    implements token bucket algorithm 
    ''' 
    client = memcache.Client(CLIENT_ARGS) 
    for _ in range(tries): 
     now = int(time.time()) 
     keys = ['%s-%s' % (key, str(now-i)) for i in range(rate_seconds)] 
     client.add(keys[0], 0, time=rate_seconds+1) 
     tokens = client.get_multi(keys[1:]) 
     tokens[keys[0]] = client.gets(keys[0]) 
     if sum(tokens.values()) >= rate_count: 
      return True 
     if client.cas(keys[0], tokens[keys[0]] + 1, time=rate_seconds+1) != 0: 
      return False 
    logging.error('cache contention error') 
    return True 

下面是用法示例：

def test_that_it_throttles_too_many_requests(self): 
    burst = 1 
    interval = 1 
    assert shared.rate_limit.throttle('test', burst, interval) is False 
    assert shared.rate_limit.throttle('test', burst, interval) is True 


def test_that_it_doesnt_throttle_burst_of_requests(self): 
    burst = 16 
    interval = 1 
    for i in range(burst): 
     assert shared.rate_limit.throttle('test', burst, interval) is False 
    time.sleep(interval + 1) # memcache has 1 second granularity 
    for i in range(burst): 
     assert shared.rate_limit.throttle('test', burst, interval) is False