Python实现简易限流器：基于信号量的并发控制详解

清欢

发布时间：2026-10-15| 5人浏览过

原创

本文介绍如何使用Python标准库中的threading.Semaphore实现简易限流器。通过设定固定数量的信号量槽位，控制任意时刻并发执行的线程数量。文章包含完整代码示例、运行效果分析及排队机制原理，帮助开发者理解基于信号量的并发限流策略。

核心原理在于动态释放资源，确保在任意给定时刻，系统中始终维持固定数量的可用信号量。

在实际开发中，我们通常采用如下方式使用信号量机制。

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。

最终，所有进站线程均完成处理并离开，等待队列清零。

本示例总共涉及102个线程：包括1个主线程、1个用于展示等候区状态的监控线程，以及代表100名进站人员的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()