亲宝软件园·资讯

展开

Python限流器

雷学委 人气:1

简单总结就是:动态的release,保证任意时刻都有固定数量可用的信号量。

我们通常会这样使用信号量

xuewei_semaphore = threading.Semaphore(4) #申请信号量

#在某个地方使用信号量
xuewei_semaphore.acquire()

//do something here
....

xuewei_semaphore.release()

限流的过程其实就是不断的使用这个有限信号量的过程。

因为设置了4信号额度,最多允许4个线程同时运行。

任意时间只要获取超过4个后,其他线程只能等待,这就跟我们进站排队很像。安检人员看到进入排队的人太多的,把后面的拦住,知道等候的人数减少,再放行一些人员进入车站等候区。

直接上代码吧,后面再解释。

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2021/11/27 10:43 下午
# @Author : LeiXueWei
# @CSDN/Juejin/Wechat: 雷学委
# @XueWeiTag: CodingDemo
# @File : threading_semephore.py
# @Project : hello
import threading
import time
import queue

xuewei_semaphore = threading.Semaphore(4)

print("xuewei_semaphore:", xuewei_semaphore)

waiting_for_train = {"value": 0}


def run():
    new_joiner = threading.current_thread().name
    # print(" %s ready" %new_joiner )
    xuewei_semaphore.acquire()
    print(" %s go" % new_joiner)
    waiting_for_train['value'] += 1
    time.sleep(1)
    print(" %s completed" % threading.current_thread().name)
    xuewei_semaphore.release()
    waiting_for_train['value'] -= 1


def log_the_waiting_area_status():
    while True:
        time.sleep(0.5)
        name = threading.current_thread().name
        print("name %s - size %s " % (name, waiting_for_train['value']))


q_watcher = threading.Thread(name="waiting area", target=log_the_waiting_area_status)
q_watcher.start()
threads = []
for i in range(100):
    t_name = "t-" + str(i)
    t = threading.Thread(name=t_name, target=run)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

这里我们申请了信号量4个空槽。

然后启动100个线程,不停的去获取信号量,然后做完就释放。

同时我们有一个缓冲队列,只存放当前新进站的人数。

通过打印这个waiting_for_train的状态,我们可以看到任意时刻队列最多只有4人进入。

也不会超过4个。

运行效果

在运行过程,我们发现queue的大小一直为4.

屏幕快照 2022-01-21 上午12.50.51.png

最后所有进站人员都进站上车了,等候的人就清零了。

屏幕快照 2022-01-21 上午12.55.15.png

这里总共有102个线程,一个主线程,一个等候区状态展示线程,还有另外一个百个线程,代表了100个进站人员。

semaphore初始化了4个度量,所以每次最多可以进站等候的人数最多只有4个。

跟地铁拦截进站一样。

我们也可以尝试把进站处理的代码修改为下方代码,读者自行运行看一下效果。

xuewei_semaphore.acquire()
print(" %s go" % new_joiner)
waiting_for_train['value'] += 1
time.sleep(1)
waiting_for_train['value'] -= 1
print(" %s completed" % threading.current_thread().name)
xuewei_semaphore.release()

总结

好,这个限流器非常简单,配套在这个中级编程简单带过一下。

读者朋友们可以把代码拷贝,运行几次,思考一下。

加载全部内容

相关教程
猜你喜欢
用户评论