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

作者:扎心了老铁 时间:2023-07-11 19:25:38 

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

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

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

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

来源:http://www.cnblogs.com/kangoroo/p/7700758.html?utm_source=tuicool&utm_medium=referral

标签:python,限流器
0
投稿

猜你喜欢

  • Python装饰器详细介绍

    2022-09-28 17:52:16
  • Python unittest装饰器实现原理及代码

    2022-05-06 22:33:49
  • SQL Server 2000中的触发器使用

    2024-01-15 15:33:34
  • Python定时执行之Timer用法示例

    2021-09-14 21:46:01
  • Python ORM框架SQLAlchemy学习笔记之数据添加和事务回滚介绍

    2023-06-15 22:47:05
  • 基于python图书馆管理系统设计实例详解

    2023-06-28 23:44:13
  • Python中字符串对齐方法介绍

    2022-07-18 06:54:50
  • 谈谈网页设计中的字体应用 (2) serif 和 sans-serif

    2009-11-24 13:04:00
  • python处理csv数据动态显示曲线实例代码

    2022-05-01 00:35:05
  • 如何利用python发送邮件

    2022-11-09 09:34:31
  • Python selenium爬取微信公众号文章代码详解

    2023-12-10 19:43:18
  • SQL Server中调用C#类中的方法实例(使用.NET程序集)

    2024-01-23 17:24:56
  • 解决 IE6 内存泄露的另类方法

    2008-07-06 23:05:00
  • 浅谈Python数据类型之间的转换

    2023-12-06 03:58:31
  • javascript中可能用得到的全部的排序算法

    2024-04-29 13:45:34
  • python中virtualenvwrapper安装与使用

    2022-07-28 03:21:52
  • Javascript中常见的逻辑题和解决方法

    2024-05-22 10:31:21
  • Oracle入侵常用操作命令整理

    2009-03-04 11:11:00
  • js实现的全国省市二级联动下拉选择菜单完整实例

    2023-09-09 05:21:01
  • 最新版MySQL 8.0.22下载安装超详细教程(Windows 64位)

    2024-01-18 15:11:02
  • asp之家 网络编程 m.aspxhome.com