Last active
November 1, 2022 15:47
-
-
Save saosir/14fdb13e0b472eff75cf18fa44c28a7e to your computer and use it in GitHub Desktop.
python令牌桶实现算法
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python | |
| # -*- encoding: utf-8 -*- | |
| # vim: set et sw=4 ts=4 sts=4 ff=unix fenc=utf8: | |
| # Author: Binux<[email protected]> | |
| # http://binux.me | |
| # Created on 2014-02-07 16:53:08 | |
| import time | |
| try: | |
| import threading as _threading | |
| except ImportError: | |
| import dummy_threading as _threading | |
| class Bucket(object): | |
| ''' | |
| traffic flow control with token bucket | |
| ''' | |
| update_interval = 30 | |
| def __init__(self, rate=1, burst=None): | |
| self.rate = float(rate) | |
| if burst is None: | |
| self.burst = float(rate) * 10 | |
| else: | |
| self.burst = float(burst) | |
| self.mutex = _threading.Lock() | |
| self.bucket = self.burst | |
| self.last_update = time.time() | |
| def get(self): | |
| '''Get the number of tokens in bucket''' | |
| now = time.time() | |
| if self.bucket >= self.burst: | |
| self.last_update = now | |
| return self.bucket | |
| bucket = self.rate * (now - self.last_update) | |
| self.mutex.acquire() | |
| if bucket > 1: | |
| self.bucket += bucket | |
| if self.bucket > self.burst: | |
| self.bucket = self.burst | |
| self.last_update = now | |
| self.mutex.release() | |
| return self.bucket | |
| def set(self, value): | |
| '''Set number of tokens in bucket''' | |
| self.bucket = value | |
| def desc(self, value=1): | |
| '''Use value tokens''' | |
| self.bucket -= value |
Author
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
rate限速流量，burst允许突增流量，每次通过get函数获取令牌桶，消耗后通过desc函数减掉消耗的令牌桶，get函数每次调用都会计算应该向向池里添加多少令牌桶