项目中用到了限流，受限于一些实现方式上的东西，手撕了一个简单的服务端限流器。
服务端限流和客户端限流的区别，简单来说就是：
1）服务端限流
对接口请求进行限流，限制的是单位时间内请求的数量，目的是通过有损来换取高可用。
例如我们的场景是，有一个服务接收请求，处理之后，将数据bulk到Elasticsearch中进行索引存储，bulk索引是一个很耗费资源的操作，如果遭遇到请求流量激增，可能会压垮Elasticsearch（队列阻塞，内存激增），所以需要对流量的峰值做一个限制。
2）客户端限流
限制的是客户端进行访问的次数。
例如，线程池就是一个天然的限流器。限制了并发个数max_connection，多了的就放到缓冲队列里排队，排队搁不下了>queue_size就扔掉。
本文是服务端限流器。
我这个限流器的优点：
1）简单
2）管事
缺点：
1）不能做到平滑限流
例如大家尝尝说的令牌桶算法和漏桶算法（我感觉这两个算法本质上都是一个事情）可以实现平滑限流。什么是平滑限流？举个栗子，我们要限制5秒钟内访问数不超过1000，平滑限流能做到，每秒200个，5秒钟不超过1000，很平衡；非平滑限流可能，在第一秒就访问了1000次，之后的4秒钟全部限制住。"htmlcode">
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就全部限制住了。也就是不能平滑限流。
不过如果你后台的逻辑有队列，或者线程池这样的缓冲，这个不平滑的影响其实不大。
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。
免责声明：本站文章均来自网站采集或用户投稿，网站不提供任何软件下载或自行开发的软件！ 如有用户或公司发现本站内容信息存在侵权行为，请邮件告知！ 858582#qq.com
《魔兽世界》大逃杀！60人新游玩模式《强袭风暴》3月21日上线
暴雪近日发布了《魔兽世界》10.2.6 更新内容，新游玩模式《强袭风暴》即将于3月21 日在亚服上线，届时玩家将前往阿拉希高地展开一场 60 人大逃杀对战。
艾泽拉斯的冒险者已经征服了艾泽拉斯的大地及遥远的彼岸。他们在对抗世界上最致命的敌人时展现出过人的手腕，并且成功阻止终结宇宙等级的威胁。当他们在为即将于《魔兽世界》资料片《地心之战》中来袭的萨拉塔斯势力做战斗准备时，他们还需要在熟悉的阿拉希高地面对一个全新的敌人──那就是彼此。在《巨龙崛起》10.2.6 更新的《强袭风暴》中，玩家将会进入一个全新的海盗主题大逃杀式限时活动，其中包含极高的风险和史诗级的奖励。
《强袭风暴》不是普通的战场，作为一个独立于主游戏之外的活动，玩家可以用大逃杀的风格来体验《魔兽世界》，不分职业、不分装备（除了你在赛局中捡到的），光是技巧和战略的强弱之分就能决定出谁才是能坚持到最后的赢家。本次活动将会开放单人和双人模式，玩家在加入海盗主题的预赛大厅区域前，可以从强袭风暴角色画面新增好友。游玩游戏将可以累计名望轨迹，《巨龙崛起》和《魔兽世界：巫妖王之怒 经典版》的玩家都可以获得奖励。