Python实现服务端限流器：单进程与分布式场景解析
本文介绍了一种基于Python的手写服务端限流器，涵盖单进程多线程与多进程分布式两种场景。通过抽象基类定义接口，分别采用线程安全队列和Redis存储实现限流逻辑。文章详细阐述了限流器的优缺点，如非平滑限流的特性，并提供了完整的代码实现与测试示例，适用于高并发场景下的流量控制。
在项目中应用了限流机制，由于现有实现方式存在局限，因此手动编写了一个简易的服务端限流器。
服务端限流与客户端限流的核心区别在于：
1）服务端限流
针对接口请求进行流量控制，限制单位时间内的请求数量，旨在通过牺牲部分可用性来保障系统的高可用。
以具体场景为例，某服务接收请求处理后，将数据批量写入Elasticsearch进行索引存储。由于bulk索引操作资源消耗巨大，若遭遇流量激增，可能导致Elasticsearch被压垮（如队列阻塞、内存激增），因此需要对流量峰值进行限制。
2）客户端限流
主要限制客户端的访问频率。
例如，线程池本身就是一种天然的限流器，它限制了最大并发数max_connection，超出部分放入缓冲队列排队，若队列容量超过queue_size则丢弃请求。
本文重点讨论服务端限流器的实现。
该限流器的优势在于：
1）实现简单 2）功能有效
不足之处包括：
1）无法实现平滑限流
常见的令牌桶算法和漏桶算法（两者本质相似）可实现平滑限流。所谓平滑限流，例如限制5秒内访问不超过1000次，平滑限流能确保每秒约200次，分布均匀；而非平滑限流可能在第一秒内处理完1000次请求，随后4秒全部拒绝。•2）灵活性不足
仅实现了秒级限流。
支持两种运行场景：
1）单进程多线程场景（使用线程安全的Queue作为全局变量）
此种场景下仅部署一个实例，对该实例进行限流，但在生产环境中较少使用。
2）多进程分布式场景（使用Redis作为全局变量）
多实例部署通常是生产环境的标准配置。
在此场景下，需对整体流量进行把控。例如user服务部署了三个实例，对外暴露query接口，目标是限制该接口的整体峰值流量，而不关心具体负载分配到哪个实例。
补充说明，此功能也可通过nginx实现。
接下来介绍限流器的具体实现。
1、接口BaseRateLimiter
按照设计思路，首先定义一个接口或抽象类。
初始化时需配置rate参数，即限流器的限速值。
提供一个抽象方法acquire()，调用该方法以判断是否限制流量。
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计数加1，若超过限速值则返回限制，否则允许访问。
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方式访问外部存储。
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抽象类定义如下：
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操作同时执行成功。
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
多线程场景下的测试代码示例：
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()
分布式场景下的测试代码示例：
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全部被限制的情况，即无法实现平滑限流。
若后台逻辑包含队列或线程池等缓冲机制，这种非平滑限流的影响通常不大。
以上内容为本文全部，希望对读者学习有所帮助，也欢迎大家支持本站。
本站声明
本文内容仅供参考，不代表本站观点。转载或引用请注明出处。如发现内容问题，请通过联系我们页面反馈。
