Python threading.Semaphore 信号量对象

说明

《Python 教程》持续更新中，提供建议、纠错、催更等加作者微信: gairuo123（备注：pandas教程）和关注公众号「盖若」ID: gairuo。跟作者学习，请进入 Python学习课程。欢迎关注作者出版的书籍：《深入浅出Pandas》和《Python之光》。

Python threading 模块中的 Semaphore 对象是一种同步原语，用于控制对共享资源的访问。它维护一个内部计数器，用来限制可以同时访问资源的线程数量。

原理

信号量是一种用于线程同步的计数器，用来限制可以访问某资源或代码块的线程数量。它允许多个线程同时访问共享资源，但对同时访问的线程数加以限制。

信号量维护一个计数器，初始值为指定的值（默认为1）。每次线程获取信号量时，计数器减1；释放信号量时，计数器加1。当计数器为0时，尝试获取信号量的线程将被阻塞，直到其他线程释放信号量。

信号量维护着一个内部计数器。当线程调用 acquire 方法时，如果计数器大于 0 ，则计数器减 1 并且线程继续执行；如果计数器为 0 ，则线程会被阻塞，直到有其他线程调用 release 方法使计数器增加，从而唤醒被阻塞的线程。

信号量对象

构造方法：

semaphore = threading.Semaphore(value=1)

创建一个信号量对象。value 指定信号量的初始计数值，默认是 1。

常用方法：

acquire(blocking=True, timeout=None): 获取信号量。如果信号量计数大于零，则计数减一并返回 True。否则，线程阻塞直到信号量计数大于零或超时。
release(): 释放信号量，计数加一，并唤醒一个阻塞的线程（如果有）。

使用场景：

资源池管理：限制同时访问资源的线程数
并发控制：限制并发执行的任务数量
生产者-消费者问题：控制缓冲区的访问
流量控制：限制API调用或数据库连接数
线程同步：协调多个线程的执行顺序

示例代码

简单的示例代码为：

import threading
import time

# 创建一个信号量对象，初始值为2
semaphore = threading.Semaphore(2)


def task(n):
    with semaphore:
        print(f"线程 {n} 获取信号量")
        time.sleep(1)  # 模拟任务
        print(f"线程 {n} 释放信号量")


threads = []
for i in range(5):
    t = threading.Thread(target=task, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

再如：

import threading
import time

semaphore = threading.Semaphore(2)  # 最多允许 2 个线程同时访问


def worker(name):
    with semaphore:
        print(f"{name} 获得资源访问权限")
        time.sleep(2)
        print(f"{name} 完成操作，释放资源")


threads = [
    threading.Thread(target=worker, args=("线程 1",)),
    threading.Thread(target=worker, args=("线程 2",)),
    threading.Thread(target=worker, args=("线程 3",)),
    threading.Thread(target=worker, args=("线程 4",))
]

for thread in threads:
    thread.start()

for thread in threads:
    thread.join()

案例

这个案例模拟了一个简单的网络爬虫，使用信号量来限制并发请求数，以避免对目标服务器造成过大压力。

import threading
import time
import random
import queue


class WebCrawler:
    def __init__(self, max_concurrent_requests=3):
        self.semaphore = threading.Semaphore(max_concurrent_requests)
        self.queue = queue.Queue()
        self.results = []
        self.result_lock = threading.Lock()

    def add_url(self, url):
        self.queue.put(url)

    def crawl(self):
        while True:
            try:
                url = self.queue.get(block=False)
            except queue.Empty:
                break

            with self.semaphore:
                result = self.fetch_url(url)
                with self.result_lock:
                    self.results.append(result)

            self.queue.task_done()

    def fetch_url(self, url):
        # 模拟网络请求
        print(f"Fetching: {url}")
        time.sleep(random.uniform(0.5, 2))
        return f"Content of {url}"

    def run(self, num_threads):
        threads = []
        for _ in range(num_threads):
            thread = threading.Thread(target=self.crawl)
            thread.start()
            threads.append(thread)

        for thread in threads:
            thread.join()

    def get_results(self):
        return self.results


def main():
    crawler = WebCrawler(max_concurrent_requests=3)

    # 添加要爬取的URL
    for i in range(10):
        crawler.add_url(f"http://example.com/page{i}")

    # 开始爬取
    start_time = time.time()
    crawler.run(num_threads=5)
    end_time = time.time()

    # 打印结果
    print("\nCrawling Results:")
    for result in crawler.get_results():
        print(result)

    print(f"\nTotal time: {end_time - start_time:.2f} seconds")
    print(f"Total pages crawled: {len(crawler.get_results())}")


if __name__ == "__main__":
    main()

这个综合案例展示了以下几点：

使用 Semaphore 来限制并发请求数，避免对服务器造成过大压力。
使用 Queue 来管理待爬取的 URL，实现任务的动态分配。
使用多线程来并发执行爬虫任务，提高效率。
使用锁（Lock）来保护共享的结果列表，确保线程安全。
模拟了网络请求的延迟，展示了信号量在控制并发访问资源时的作用。
展示了如何结合使用多个同步原语（Semaphore, Lock）来解决复杂的并发问题。

这个例子很好地说明了 Semaphore 在实际应用中的重要性：

资源限制：通过限制并发请求数，避免了对目标服务器的过度负载。
性能优化：允许多个线程并发工作，但又不至于无限制地创建线程或发送请求。
灵活性：可以通过调整 max_concurrent_requests 参数来平衡性能和资源使用。
可扩展性：这种设计可以轻松扩展到更复杂的爬虫系统，如添加错误处理、重试机制等。

Semaphore 在需要控制并发访问资源数量的场景中非常有用。它提供了一种简单而有效的方式来管理共享资源，特别适合在网络编程、数据库连接池、线程池等领域使用。

注意事项

使用 Semaphore 时需要注意：

正确设置初始值，既要保证并发性，又要避免资源耗尽。
确保每次 acquire 操作都有对应的 release 操作，避免死锁。
考虑使用 with 语句来自动管理信号量的获取和释放。

在更复杂的系统中，Semaphore 常常与其他同步原语（如 Lock, Condition）结合使用，以实现更复杂的并发控制策略。

参考

https://docs.python.org/zh-cn/3/library/threading.html#semaphore-objects

< Condition 线程条件对象 Python 模块 threading 基于线程的并行 Event 事件对象 >

更新时间：June 30, 2024, 5:28 p.m. 标签：python threading 线程信号量对象