Token Bucket在QoS中入门级介绍python示例

概述TokenBucket为令牌桶算法，常被用于流量整形，或对客户端的限速。假设家里是200M的宽带，为了保证打游戏的流畅，我要把电视设置为10M。那么实现上可以如下： 方案一：累加电视的所有接收/发出的数据包长度，如果超出10M，则丢弃报文；每秒更新一次累加值为0。思路非常简单，也非常实用；从长时间统计

Token Bucket为令牌桶算法，常被用于流量整形，或对客户端的限速。

假设家里是200M的宽带， 为了保证打游戏的流畅，我要把电视设置为10M。那么实现上可以如下：

 

方案一：

累加电视的所有接收/发出的数据包长度，如果超出10M，则丢弃报文；每秒更新一次累加值为0。

思路非常简单，也非常实用；从长时间统计上看，结果上偏差不大。

但可能限制出来流量图就会如下，不均匀，像抽风似的。如果设置了10条限速规则对10个设备进行限速时，则每秒开始的0.1秒，cpu压力很大，然后0.9秒很闲。

 

这种漏桶似的算法，利用率和平衡性较差。

方案二：

如果改进一下，将统计时间细化0.1秒，将效果会好不少。再进一步，将时间细化到每个报文接收，则效果就达到极致。

这就是Token Bucket做的事情。

 

流程出下：

1、当收到数据报文时，先试图获取token，如果剩余token足够则放行；

2、不够则更新token，等待token足够再继续。token按指定速率添加到bucket中，超过其容量则设置为最大值。

预期效果图如下：

 

示例代码实现如下：假设允许的速率每秒100，模拟包长在50~100随机，统计一下发送时间和平均速率。

 

# -- Coding: utf-8 --# Good luck.import timeimport randomclass tb():    def __init__(self, token_per_sec):        self.captoken = token_per_sec        self.curr_token = 0        self.timestamp = time.time()    def refill(self):        if time.time() > self.timestamp:            refill_token = self.captoken*(time.time() - self.timestamp)            self.curr_token = min(self.curr_token + refill_token, self.captoken)            print "\tRefill tokens Current: %d. (Add %d)" %(self.curr_token, refill_token)            self.timestamp = time.time()        else:            print "\tNo refill...."    def consume(self, num):        ret = False        if self.curr_token > 0: # Control precision            self.curr_token -=  num            print "\tGet token %d : remain %d" %(num, self.curr_token)            ret = True        self.refill()        return rettest = tb(100)run_time = time.time()all_len = 0;times = 0while True:    t = 0.1    ts = time.time()    send_len = random.randint(10, 50)    print "Need send %d" %(send_len)    while not test.consume(send_len):        time.sleep(t)    te = time.time()    print "Use Time:%s Send:%d " %(te-ts, send_len)    all_len += send_len    times += 1    print "Avg = %s (times=%d)" %(all_len/(ts - run_time+0.1), times)    time.sleep(t)

运行效果

C:\Python27\python.exe X:/GIT/wxy_code_backup/Flash_tools/tokenbucket.py

Need send 44

No refill....

Refill tokens Current: 9. (Add 9)

Get token 44 : remain -34

Refill tokens Current: -23. (Add 10)

Use Time:0.200999975204 Send:44

Avg = 440.0 (times=1)

Need send 27

Refill tokens Current: -13. (Add 9)

Refill tokens Current: -3. (Add 10)

Refill tokens Current: 6. (Add 10)

Get token 27 : remain -20

Refill tokens Current: -10. (Add 10)

Use Time:0.302000045776 Send:27

Avg = 177.057409665 (times=2)

Need send 33

Refill tokens Current: 0. (Add 10)

Refill tokens Current: 9. (Add 10)

Get token 33 : remain -23

Refill tokens Current: -13. (Add 10)

Use Time:0.200999975204 Send:33

Avg = 129.51431022 (times=3)

Need send 42

Refill tokens Current: -3. (Add 10)

Refill tokens Current: 6. (Add 10)

Get token 42 : remain -35

Refill tokens Current: -25. (Add 10)

Use Time:0.201000213623 Send:42

Avg = 132.126711657 (times=4)

Need send 16

Refill tokens Current: -15. (Add 9)

Refill tokens Current: -5. (Add 9)

Refill tokens Current: 4. (Add 10)

Get token 16 : remain -11

Refill tokens Current: -1. (Add 9)

Use Time:0.300999879837 Send:16

Avg = 115.22048411 (times=5)

......

Need send 21

Refill tokens Current: -4. (Add 10)

Refill tokens Current: 5. (Add 9)

Get token 21 : remain -15

Refill tokens Current: -5. (Add 10)

Use Time:0.200999975204 Send:21

Avg = 100.389763573 (times=138)

Need send 41

Refill tokens Current: 4. (Add 9)

Get token 41 : remain -36

Refill tokens Current: -26. (Add 10)

Use Time:0.101000070572 Send:41

Avg = 100.66413934 (times=139)

Need send 18

Refill tokens Current: -15. (Add 10)

Refill tokens Current: -5. (Add 10)

Refill tokens Current: 4. (Add 10)

Get token 18 : remain -13

Refill tokens Current: -3. (Add 10)

Use Time:0.301999807358 Send:18

Avg = 100.607594694 (times=140)

 

稳定后效果还算可以，从发包点上也相对均匀。当然初始仅为波动而已，可以通过调整优化掉，但从长时间看，效果是一致的。

 

方案三：

方案一、二都缺少缓冲的支持。如果报文来了，但是现在资源不够，不能发送，直接丢弃，则客户端再重发，这个过程太浪费。

因此增加缓冲队列，根据系统的内存，设定队列的大小，对于不能发送的报文，先入队。

如果队列满了呢？ 那只能丢弃了，毫无办法。（不考虑共享宽带的情况）。

从linux收发包角度，每个以太网设备有一个发送队列，如果满了，则设置为状态不可用，则不再收包。道理是一样的。

 

关于更新Token时间的设计，如上例程序，仅为演示，我没有加线程额外处理。

在存在缓冲队列的情况下，则必须增加更新线程，因为存在收发异常的情况。

 

示例如下，不过缓冲不过是增加一些d性，丢包依然不可避免。结果可预知：由于缓冲的存在，平均流量会在一段时间后，才能趋向限制值。

# -- Coding: utf-8 --# Good luck.import timeimport randomimport threadingclass tb():    def __init__(self, token_per_sec):        self.captoken = token_per_sec        self.curr_token = 0        self.timestamp = time.time()        self.queue = []        self.q_max = 5    def refill(self):        if time.time() > self.timestamp:            refill_token = self.captoken*(time.time() - self.timestamp)            self.curr_token = min(self.curr_token + refill_token, self.captoken)            print "\tRefill tokens Current: %d. (Add %d)" %(self.curr_token, refill_token)            self.timestamp = time.time()        else:            print "\tNo refill...."    def consume(self, num):        ret = False        if self.curr_token > 0: # Control precision            self.curr_token -=  num            print "\tGet token %d : remain %d" %(num, self.curr_token)            ret = True        elif len(self.queue) < self.q_max:            # Enter queue            self.queue.append(num)            print "Enqueue Action."            ret = True        else:            # Drop it.            ret = False        self.refill()        return ret    def dequeue(self):        if len(self.queue) and self.curr_token > 0:            tmp = self.queue.pop()            self.curr_token -= tmp            print "Dequeue Action : %d (remain %d)" %(tmp, self.curr_token)def update_bucket(args):    print "Args=%s" %(args.captoken)    while True:        time.sleep(0.1)        args.refill()        args.dequeue()test = tb(100)run_time = time.time()all_len = 0times = 0tx_queue = []# update bucket thread.update_t = threading.Thread(target=update_bucket, args=(test,))update_t.start()while True:    t = 0.2    ts = time.time()    send_len = random.randint(40, 50)    print "Need send %d" %(send_len)    if test.consume(send_len):        te = time.time()        print "Use Time:%s Send:%d " %(te-ts, send_len)        all_len += send_len    else:        print "Drop Action."    times += 1    print "Avg = %s (times=%d)" %(all_len/(ts - run_time+0.1), times)    time.sleep(t)

运行结果：

Args=100

Need send 46

Enqueue Action.

Refill tokens Current: 0. (Add 0)

Use Time:0.0 Send:46

Avg = 455.44587139 (times=1)

Refill tokens Current: 10. (Add 9)

Dequeue Action : 46 (remain -35)

Need send 41

Enqueue Action.

Refill tokens Current: -25. (Add 10)

Use Time:0.0 Send:41

Avg = 289.036568661 (times=2)

Refill tokens Current: -25. (Add 0)

Refill tokens Current: -15. (Add 10)

Need send 47

Enqueue Action.

Refill tokens Current: -5. (Add 9)

Use Time:0.0 Send:47

Avg = 266.932297286 (times=3)

Refill tokens Current: -5. (Add 0)

Refill tokens Current: 4. (Add 10)

Dequeue Action : 47 (remain -42)

 

......

Drop Action.

Avg = 106.217083485 (times=170)

Refill tokens Current: -9. (Add 2)

Refill tokens Current: 0. (Add 10)

Dequeue Action : 42 (remain -41)

Need send 50

Enqueue Action.

Refill tokens Current: -34. (Add 7)

Use Time:0.0 Send:50

Avg = 107.055107617 (times=171)

Refill tokens Current: -31. (Add 2)

 

在有针对多个用户的多个限速规则，必然存在多个队列后，就需要有一定的调度算法，常用的RR、DRR、WRR，后面使用一些示例，展开讨论一下。

 

 

 

 

 

 

总结

以上是内存溢出为你收集整理的Token Bucket在QoS中入门级介绍python示例全部内容，希望文章能够帮你解决Token Bucket在QoS中入门级介绍python示例所遇到的程序开发问题。

如果觉得内存溢出网站内容还不错，欢迎将内存溢出网站推荐给程序员好友。