Threading.Lock.acquire: примеры (PYTHON)

Примеры использования threading.Lock.acquire

Стандартное использование для защиты общего ресурса:

import threading
counter = 0
lock = threading.Lock()

def increment():
    global counter
    lock.acquire()
    try:
        counter += 1
    finally:
        lock.release()

threads = []
for _ in range(1000):
    t = threading.Thread(target=increment)
    threads.append(t)
    t.start()
for t in threads:
    t.join()
print(f'Итоговое значение счетчика: {counter}')

Итоговое значение счетчика: 1000

Использование с аргументом blocking=False:

import threading
lock = threading.Lock()
lock.acquire()
print(f'Первая попытка (blocking=False): {lock.acquire(blocking=False)}')
print(f'Вторая попытка (blocking=False): {lock.acquire(False)}')
lock.release()
print(f'Третья попытка после release (blocking=False): {lock.acquire(blocking=False)}')

Первая попытка (blocking=False): False
Вторая попытка (blocking=False): False
Третья попытка после release (blocking=False): True

Использование с аргументом timeout:

import threading, time
lock = threading.Lock()
lock.acquire()

def try_lock(timeout_val):
    result = lock.acquire(timeout=timeout_val)
    print(f'Попытка получить блокировку с timeout={timeout_val}: {result}')
    if result:
        lock.release()

t = threading.Thread(target=try_lock, args=(0.5,))
t.start()
time.sleep(1.0)  # Держим блокировку дольше, чем timeout в потоке
lock.release()
t.join()

Попытка получить блокировку с timeout=0.5: False

Похожие функции в Python

В модуле threading существуют другие примитивы синхронизации:

• threading.RLock (Reentrant Lock): Позволяет одному потоку многократно получать одну и ту же блокировку без взаимной блокировки. Каждому вызову acquire() должен соответствовать вызов release(). Используется, когда функция, владеющая блокировкой, вызывает другую функцию, которой также требуется эта блокировка.
• with lock: Контекстный менеджер, который автоматически вызывает acquire() при входе в блок и release() при выходе. Это предпочтительный способ работы с блокировками, так как он гарантирует освобождение ресурса даже при возникновении исключения.
• threading.Semaphore: Счеточный семафор, позволяющий одновременно работать с ресурсом заданному количеству потоков. Метод acquire() уменьшает внутренний счетчик.
• threading.BoundedSemaphore: Семафор, который не позволяет счетчику превысить начальное значение.
• threading.Condition: Объединяет блокировку и механизм уведомлений, позволяя потокам ждать определенного условия. Метод acquire() у объекта Condition наследуется от связанной блокировки.

Обычный Lock предпочтителен для простых сценариев защиты критического участка. RLock выбирается при необходимости рекурсивного захвата. Семафоры применяются для ограничения доступа к ресурсу с ограниченной емкостью.

Типичные ошибки при использовании

1. Отсутствие вызова release() после acquire(), приводящее к взаимной блокировке (deadlock).

import threading
lock = threading.Lock()
lock.acquire()
# Забыли release()
# Дальнейшие вызовы acquire() будут ждать вечно
print("Эта строка будет напечатана")
# В другом потоке или позже в этом же:
lock.acquire()  # Вечная блокировка здесь
print("Эта строка никогда не будет напечатана")

2. Попытка освободить блокировку, которая не была захвачена текущим потоком, вызывает RuntimeError.

import threading
lock = threading.Lock()
try:
    lock.release()
except RuntimeError as e:
    print(f'Ошибка: {e}')

Ошибка: release unlocked lock

3. Захват блокировки обычным Lock дважды в одном потоке без промежуточного освобождения приводит к deadlock.

import threading
lock = threading.Lock()
lock.acquire()
lock.acquire()  # Вечная блокировка здесь
print("Сюда выполнение не дойдет")

4. Использование разных объектов блокировки для защиты одного и того же ресурса.

import threading
counter = 0
lock1 = threading.Lock()
lock2 = threading.Lock()  # Ошибка: должна быть одна блокировка

def increment():
    global counter
    lock1.acquire()  # Используем lock1
    counter += 1
    lock1.release()

def decrement():
    global counter
    lock2.acquire()  # Используем lock2, ошибка!
    counter -= 1
    lock2.release()

Для избежания этих ошибок рекомендуется использовать контекстный менеджер (with lock:).

Изменения в последних версиях Python

В Python 3.2 была изменена внутренняя реализация модуля threading для улучшения производительности и надежности, но публичный API Lock.acquire остался прежним.

Начиная с Python 3.10, в документации явно указано, что блокировки, реализованные на C, считаются атомарными с точки зрения сигналов. В Python 3.13 планируется улучшение производительности примитивов синхронизации.

Важное изменение, касающееся поведения: в Python 3.9 и более ранних версиях аргумент timeout мог быть равен None для неограниченного ожидания. Начиная с Python 3.10, значение None явно эквивалентно -1 для обратной совместимости, но рекомендуется использовать -1 для ясности.

Сам метод acquire не претерпевал значительных изменений в сигнатуре с момента введения параметра timeout в ранних версиях Python 2.x.

Расширенные примеры использования

Использование блокировки для реализации семафора с ограничением в N одновременных доступов:

import threading, time
class SimpleSemaphore:
    def __init__(self, value):
        self.value = value
        self.lock = threading.Lock()
        self.condition = threading.Condition(self.lock)
    
    def acquire(self):
        with self.condition:
            while self.value == 0:
                self.condition.wait()
            self.value -= 1
    
    def release(self):
        with self.condition:
            self.value += 1
            self.condition.notify()

sem = SimpleSemaphore(3)  # Одновременно 3 потока
counter = 0
counter_lock = threading.Lock()

def worker(id):
    global counter
    sem.acquire()
    try:
        with counter_lock:
            counter += 1
            print(f'Поток {id} вошел, счетчик: {counter}')
        time.sleep(1)
        with counter_lock:
            counter -= 1
            print(f'Поток {id} вышел, счетчик: {counter}')
    finally:
        sem.release()

threads = [threading.Thread(target=worker, args=(i,)) for i in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

Поток 0 вошел, счетчик: 1
Поток 1 вошел, счетчик: 2
Поток 2 вошел, счетчик: 3
Поток 1 вышел, счетчик: 2
Поток 0 вышел, счетчик: 1
Поток 2 вышел, счетчик: 0
Поток 3 вошел, счетчик: 1
Поток 4 вошел, счетчик: 2
... и т.д.

Потокобезопасный кэш с использованием блокировки и проверкой двойного захвата:

import threading
class ThreadSafeCache:
    def __init__(self):
        self.cache = {}
        self.lock = threading.Lock()
    
    def get_or_compute(self, key, compute_func):
        # Первая проверка без блокировки (для скорости)
        if key in self.cache:
            return self.cache[key]
        
        # Получаем блокировку
        self.lock.acquire()
        try:
            # Вторая проверка под блокировкой (double-checked locking)
            if key in self.cache:
                return self.cache[key]
            # Вычисляем значение
            value = compute_func()
            self.cache[key] = value
            return value
        finally:
            self.lock.release()

cache = ThreadSafeCache()
import random
def expensive_computation():
    time.sleep(0.5)  # Имитация долгого вычисления
    return random.randint(1, 100)

def worker(key):
    value = cache.get_or_compute(key, expensive_computation)
    print(f'Key {key}: {value}')

keys = ['a', 'b', 'a', 'c', 'b']
threads = []
for i, key in enumerate(keys):
    t = threading.Thread(target=worker, args=(key,))
    threads.append(t)
    t.start()
for t in threads:
    t.join()

Использование блокировки с таймаутом для предотвращения deadlock в распределенных сценариях:

import threading, time
lock_a = threading.Lock()
lock_b = threading.Lock()

def thread_1():
    print('Поток 1: пытаюсь захватить lock_a')
    if lock_a.acquire(timeout=1):
        try:
            print('Поток 1: захватил lock_a, сплю 0.5 сек')
            time.sleep(0.5)
            print('Поток 1: пытаюсь захватить lock_b с таймаутом 0.5 сек')
            if lock_b.acquire(timeout=0.5):
                try:
                    print('Поток 1: захватил lock_b')
                finally:
                    lock_b.release()
            else:
                print('Поток 1: не удалось захватить lock_b (таймаут)')
        finally:
            lock_a.release()
    else:
        print('Поток 1: не удалось захватить lock_a (таймаут)')

def thread_2():
    print('Поток 2: пытаюсь захватить lock_b')
    if lock_b.acquire(timeout=1):
        try:
            print('Поток 2: захватил lock_b, сплю 0.5 сек')
            time.sleep(0.5)
            print('Поток 2: пытаюсь захватить lock_a с таймаутом 0.5 сек')
            if lock_a.acquire(timeout=0.5):
                try:
                    print('Поток 2: захватил lock_a')
                finally:
                    lock_a.release()
            else:
                print('Поток 2: не удалось захватить lock_a (таймаут)')
        finally:
            lock_b.release()
    else:
        print('Поток 2: не удалось захватить lock_b (таймаут)')

th1 = threading.Thread(target=thread_1)
th2 = threading.Thread(target=thread_2)
th1.start()
th2.start()
th1.join()
th2.join()
print('Оба потока завершились без deadlock')

Поток 1: пытаюсь захватить lock_a
Поток 2: пытаюсь захватить lock_b
Поток 1: захватил lock_a, сплю 0.5 сек
Поток 2: захватил lock_b, сплю 0.5 сек
Поток 1: пытаюсь захватить lock_b с таймаутом 0.5 сек
Поток 2: пытаюсь захватить lock_a с таймаутом 0.5 сек
Поток 1: не удалось захватить lock_b (таймаут)
Поток 2: не удалось захватить lock_a (таймаут)
Оба потока завершились без deadlock

Аналоги функции в других языках программирования

Java: Используется интерфейс Lock и его реализация ReentrantLock. Метод tryLock() аналогичен acquire(blocking=False), а lock() - acquire().

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
Lock lock = new ReentrantLock();
if (lock.tryLock()) {
    try {
        // Критическая секция
    } finally {
        lock.unlock();
    }
}

JavaScript (Node.js): Нет встроенных примитивов для потоков (worker threads используют изолированную память). Для асинхронного кода часто применяются другие механизмы (мьютексы из библиотек, атомарные операции Atomics).

Go: Используются мьютексы из пакета sync. Метод Lock() блокирующий, а TryLock() (с версии Go 1.18) возвращает булево значение.

import "sync"
var mu sync.Mutex
mu.Lock()
// Критическая секция
mu.Unlock()
// Немедленная попытка:
if mu.TryLock() {
    defer mu.Unlock()
    // Критическая секция
}

C#: Ключевое слово lock или класс Monitor с методами Enter (аналог acquire) и TryEnter (аналог acquire с timeout или blocking=false).

object lockObj = new object();
lock (lockObj) {
    // Критическая секция
}
// Альтернативно:
if (Monitor.TryEnter(lockObj, 500)) { // timeout 500 ms
    try {
        // Критическая секция
    } finally {
        Monitor.Exit(lockObj);
    }
}

Kotlin: Используются мьютексы из корутин (Mutex). Метод lock является suspend-функцией, а tryLock возвращает булево значение.

import kotlinx.coroutines.sync.Mutex
val mutex = Mutex()
suspend fun example() {
    mutex.lock()
    try {
        // Критическая секция
    } finally {
        mutex.unlock()
    }
    // Немедленная попытка:
    if (mutex.tryLock()) {
        try {
            // Критическая секция
        } finally {
            mutex.unlock()
        }
    }
}

Основное отличие Python - наличие GIL (Global Interpreter Lock), который ограничивает выполнение байт-кода Python одним потоком в один момент времени. Однако threading.Lock все равно необходим для предотвращения race condition при переключении потоков и для синхронизации операций, отпускающих GIL (например, ввод-вывод или операции в расширениях C).