Python:有没有一种线程安全的方法来知道lock.acquire()是否已阻塞(并继续阻塞)？

Question

我正在为我的 REST 服务编写一个 API 库。在某些时候，访问 token 将需要更新。我正在尝试实现一种线程安全的方法来执行此操作，以便仅发送一个更新请求，即使多个线程可能想要同时更新它。

这是我的代码的简化版本:

import requests

class Api():
    def _renew(self):
        # send a request which renews token in self._headers

    def do_something(self, url, params=None):
        r = requests(url, params=params, headers=self._headers)
        if r.status_code == 401 and r.json()['error'] == 'Token expired':
             # renew the access token
             self._renew()
             # repeat request with updated headers
             r = requests(url, params=params, headers=self._headers)
        return r.json()

我需要知道当前的续订请求是否正在进行中。我的想法是像这样编写更新函数:

def _renew(self):
    lock.acquire()
    # i want to check here if the lock.acquire() call blocked
    # the thread and return immediately if it has
    try:
        # send a request to renew token header in self._headers
    finally:
        lock.release()

我希望其他可能调用 do_something() (以及随后的 _renew())方法的线程等待，直到第一个真正更新 token 并让其他线程使用它的结果.

如何判断我的 lock.acquire() 调用是否被阻塞？

在调用 acquire() 之前检查 lock.locked() 的结果不是线程安全的，有时会有多个线程向服务器发送更新请求。

Answer 1

您可以调用lock.acquire(False)进行非阻塞调用，并使用返回值来确定是否已获取锁。这看起来像这样:

def _renew(self):
    # calling when the lock is already acquired
    # will not block and return False
    if not lock.acquire(False):
        event.wait()
        return
    # else we acquired the lock 
    event.clear()
    try:
        # send a request to renew token header in self._headers
    finally:
        event.set()
        lock.release()

参见the threading-docs对于Python。

另一种方法是将 token 包装在另一个类中:

from threading import Event, RLock

class Token:
    _internal = RLock()
    _marker = False
    def __init__(self):
        # TODO set header
        self._header = None

    def _renew(self):
        # TODO refresh the header
        pass

    def get(self):
        with self._internal:
            if self._marker:
                self._renew()
                self._marker = False
            return self._header

    # Marks the header to be refreshed at the next get()
    def set_renew(self):
        with self._internal:
            self._marker = True

这有几个优点。首先，代币要对自己负责。在最好的环境中，它只会在需要时刷新自己，而不会被其他类刷新。这应该在 Token#get(self) 中决定。这还通过将所有 get 调用包装到锁中来解决线程安全问题，防止不必要的修改。