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

更新时间：2017年10月26日 14:39:54 作者：扎心了老铁

本篇文章主要介绍了python分布式环境下的限流器的示例，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧

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

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

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

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一次，如果大于限速了，就返回限制。否则就允许访问。

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就全部限制住了。也就是不能平滑限流。

不过如果你后台的逻辑有队列，或者线程池这样的缓冲，这个不平滑的影响其实不大。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持脚本之家。

您可能感兴趣的文章:

Python实现简易的限流器介绍

详解10个可以快速用Python进行数据分析的小技巧
这篇文章主要介绍了详解10个可以快速用Python进行数据分析的小技巧，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2019-06-06
python创建只读属性对象的方法(ReadOnlyObject)
有时需要创建一个带只读属性的对象，大家可以参考下如下的方法进行创建，稍加改造，可以得到很特殊的效果
2013-02-02
Python Concurrent Futures解锁并行化编程的魔法示例
Python的concurrent.futures模块为并行化编程提供了强大的工具,使得开发者能够轻松地利用多核心和异步执行的能力,本文将深入探讨concurrent.futures的各个方面,从基础概念到高级用法,为读者提供全面的了解和实用的示例代码
2023-12-12
用python实现名片管理系统
这篇文章主要为大家详细介绍了用python实现名片管理系统，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2020-06-06
python中使用矢量化替换循环详解
矢量化是在数据集上实现 (NumPy) 数组操作的技术。在后台，它将操作一次性应用于数组或系列的所有元素（不同于一次操作一行的“for”循环）。
2023-01-01
python 实现生成均匀分布的点
今天小编就为大家分享一篇python 实现生成均匀分布的点，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2019-12-12
python树的同构学习笔记
在本篇文章里小编给大家整理的是一篇关于python树的同构学习笔记以及相关实例代码内容，有需要的朋友们学习下。
2019-09-09
python Matplotlib数据可视化（2）：详解三大容器对象与常用设置
这篇文章主要介绍了python Matplotlib三大容器对象与常用设置的相关资料，帮助大家更好的学习和使用Matplotlib库的用法，感兴趣的朋友可以了解下
2020-09-09
Python Opencv中用compareHist函数进行直方图比较对比图片
这篇文章主要介绍了Python Opencv中用compareHist函数进行直方图比较进行对比图片，本文通过实例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
2020-04-04
Python使用pickle模块储存对象操作示例
这篇文章主要介绍了Python使用pickle模块储存对象操作,结合实例形式分析了Python使用pickle模块针对文件读写与转换的相关操作技巧,需要的朋友可以参考下
2018-08-08