快捷導(dǎo)航

Python實現(xiàn)簡易的限流器介紹

更新時間：2022年01月28日 11:07:17 作者：雷學(xué)委

大家好，本篇文章主要講的是Python實現(xiàn)簡易的限流器介紹，感興趣的同學(xué)趕快來看一看吧，對你有幫助的話記得收藏一下

簡單總結(jié)就是：動態(tài)的release，保證任意時刻都有固定數(shù)量可用的信號量。

我們通常會這樣使用信號量

xuewei_semaphore = threading.Semaphore(4) #申請信號量

#在某個地方使用信號量
xuewei_semaphore.acquire()

//do something here
....

xuewei_semaphore.release()

限流的過程其實就是不斷的使用這個有限信號量的過程。

因為設(shè)置了4信號額度，最多允許4個線程同時運行。

任意時間只要獲取超過4個后，其他線程只能等待，這就跟我們進站排隊很像。安檢人員看到進入排隊的人太多的，把后面的攔住，知道等候的人數(shù)減少，再放行一些人員進入車站等候區(qū)。

直接上代碼吧，后面再解釋。

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2021/11/27 10:43 下午
# @Author : LeiXueWei
# @CSDN/Juejin/Wechat: 雷學(xué)委
# @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個線程，不停的去獲取信號量，然后做完就釋放。

同時我們有一個緩沖隊列，只存放當(dāng)前新進站的人數(shù)。

通過打印這個waiting_for_train的狀態(tài)，我們可以看到任意時刻隊列最多只有4人進入。

也不會超過4個。

運行效果

在運行過程，我們發(fā)現(xiàn)queue的大小一直為4.

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

最后所有進站人員都進站上車了，等候的人就清零了。

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

這里總共有102個線程，一個主線程，一個等候區(qū)狀態(tài)展示線程，還有另外一個百個線程，代表了100個進站人員。

semaphore初始化了4個度量，所以每次最多可以進站等候的人數(shù)最多只有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()