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

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

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

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

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

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

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

//do something here
....

xuewei_semaphore.release()

限流的過程其實(shí)就是不斷的使用這個(gè)有限信號量的過程。

因?yàn)樵O(shè)置了4信號額度，最多允許4個(gè)線程同時(shí)運(yùn)行。

任意時(shí)間只要獲取超過4個(gè)后，其他線程只能等待，這就跟我們進(jìn)站排隊(duì)很像。安檢人員看到進(jìn)入排隊(duì)的人太多的，把后面的攔住，知道等候的人數(shù)減少，再放行一些人員進(jìn)入車站等候區(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個(gè)空槽。

然后啟動100個(gè)線程，不停的去獲取信號量，然后做完就釋放。

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

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

也不會超過4個(gè)。

運(yùn)行效果

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

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

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

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

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

semaphore初始化了4個(gè)度量，所以每次最多可以進(jìn)站等候的人數(shù)最多只有4個(gè)。

跟地鐵攔截進(jìn)站一樣。

我們也可以嘗試把進(jìn)站處理的代碼修改為下方代碼，讀者自行運(yùn)行看一下效果。

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()