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就全部限制住了。也就是不能平滑限流。
不过如果你后台的逻辑有队列,或者线程池这样的缓冲,这个不平滑的影响其实不大。
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持脚本之家。
- 【编程题】Java编程题一(10道)
- 【学习笔记】springboot教程(1)第一个demo
- Hyperledger Fabric 积分代币上链方案
- MySQL中的事务和锁简单测试(r10笔记第46天)
- 数据库无响应问题的紧急处理和分析 (r10笔记第42天)
- 【转载】记Golang数据库查询封装的坑
- Golang 基本语法学习笔记之流程控制
- Go 语言的演化历程
- JS 评分五角星随鼠标移动显示
- Golang标准库学习——buffio包 ---转
- 【Go 语言社区】Go语言条件变量的两个例子
- mysqlimport导入报错的排查(r10笔记第58天)
- 【Go 语言社区】POJ 1047 Round and Round We Go 循环数新解
- 【Go 语言社区】删除redis所有KEY
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法