python分布式环境下的限流器的示例

2023-09-22 12:30:05 161

项目中用到了限流，受限于一些实现方式上的东西，手撕了一个简单的服务端限流器。

服务端限流和客户端限流的区别，简单来说就是：

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()，调用这个方法，返回是否限制流量。

classBaseRateLimiter(object):

__metaclass__=abc.ABCMeta

@abc.abstractmethod
def__init__(self,rate):
self.rate=rate

@abc.abstractmethod
defacquire(self,count):
return

2、单进程多线程场景的限流ThreadingRateLimiter

继承BaseRateLimiter抽象类，使用线程安全的Queue作为全局变量，来消除竞态影响。

后台有个进程每秒钟清空一次queue；

当请求来了，调用acquire函数，queueincr一次，如果大于限速了，就返回限制。否则就允许访问。

classThreadingRateLimiter(BaseRateLimiter):

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

defacquire(self,count=1):
self.queue.put(1,block=False)
returnself.queue.qsize()
2、分布式场景下的限流DistributeRateLimiter
继承BaseRateLimiter抽象类，使用外部存储作为共享变量，外部存储的访问方式为cache。
classDistributeRateLimiter(BaseRateLimiter):

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

defacquire(self,count=1,expire=3,key=None,callback=None):
try:
ifisinstance(self.cache,Cache):
returnself.cache.fetchToken(rate=self.rate,count=count,expire=expire,key=key)
exceptException,ex:
returnTrue

为了解耦和灵活性，我们实现了Cache类。提供一个抽象方法getToken()
如果你使用redis的话，你就继承Cache抽象类，实现通过redis获取令牌的方法。
如果使用mysql的话，你就继承Cache抽象类，实现通过mysql获取令牌的方法。
cache抽象类
classCache(object):

__metaclass__=abc.ABCMeta

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

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

给出一个redis的实现RedisTokenCache
每秒钟创建一个key，并且对请求进行计数incr，当这一秒的计数值已经超过了限速rate，就拿不到token了，也就是限制流量。
对每秒钟创建出的key，让他超时expire。保证key不会持续占用存储空间。
没有什么难点，这里使用redis事务，保证incr和expire能同时执行成功。
classRedisTokenCache(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
))

deffetchToken(self,rate=100,count=1,expire=3,key=None):
date=datetime.now().strftime("%Y-%m-%d%H:%M:%S")
key=":".join([self.namespace,keyifkeyelseself.key,date])
try:
current=self.redis.get(key)
ifint(currentifcurrentelse"0")>rate:
raiseException("tomanyrequestsincurrentsecond:%s"%date)
else:
withself.redis.pipeline()asp:
p.multi()
p.incr(key,count)
p.expire(key,int(expireifexpireelse"3"))
p.execute()
returnTrue
exceptException,ex:
returnFalse


多线程场景下测试代码 
limiter=ThreadingRateLimiter(rate=10000)

defjob():
while1:
ifnotlimiter.acquire():
print'限流'
else:
print'正常'

threads=[threading.Thread(target=job)foriinrange(10)]
forthreadinthreads:
thread.start()

分布式场景下测试代码
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'))

defjob():
while1:
ifnotlimiter.acquire():
print'限流'
else:
print'正常'

threads=[multiprocessing.Process(target=job)foriinrange(10)]
forthreadinthreads:
thread.start()

可以自行跑一下。 
说明：
我这里的限速都是秒级别的，例如限制每秒400次请求。有可能出现这一秒的前100ms，就来了400次请求，后900ms就全部限制住了。也就是不能平滑限流。
不过如果你后台的逻辑有队列，或者线程池这样的缓冲，这个不平滑的影响其实不大。
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。

python分布式环境下的限流器的示例

热门推荐

随机推荐