механизм синхронизации потоков Python

Эта статья начинается с причин синхронизации потоков и углубляется в различные механизмы синхронизации, в основном включая следующее содержание.

• Блокировки потоков (синхронизация потоков, блокировка мьютексов)
• ГИЛ замок
• тупик
• RLock (рекурсивная блокировка, повторная блокировка)

Блокировки потоков (синхронизация потоков, блокировка мьютексов)

В многопроцессорности каждый процесс имеет копию данных, и каждый поток многопоточности использует одни и те же данные. Это позволяет многопоточности потреблять меньше ресурсов, но смешивание ресурсов приведет к некоторым ошибкам, давайте рассмотрим следующий пример.

import threading
import time
zero = 0
def change_zero():
    global zero
    for i in range(1000000):
        zero = zero + 1
        zero = zero - 1
th1 = threading.Thread(target = change_zero)
th2 = threading.Thread(target = change_zero)
th1.start()
th2.start()
th1.join()
th2.join()
print(zero)

change_zeroфункция будетzeroПеременная увеличивается на 1, а затем уменьшается на 1. Само собой разумеется, что независимо от того, сколько раз она запускается,zeroВсе переменные должны быть равны 0, но если приведенный выше код запускается много раз, всегда будут случаи, когда он не равен 0 (если цикл изменить так, чтобы он выполнялся 10 000 000 раз, результат будет трудно получить 0), указывая на то, что разные потоки модифицируются вместе.zeroПеременная перепутана, давайте посмотрим на причину путаницы. (Ссылаться наздесь)

zero = zero + 1В python сначала будет сгенерирована промежуточная переменная, напримерx1 = zero + 1,после этогоzero = x1.

Если многопоточность не используется, запустите дваждыchange_zeroФункция такая

初始：zero = 0
th1: x1 = zero + 1  # x1 = 1
th1: zero = x1      # zero = 1
th1: x1 = zero - 1  # x1 = 0
th1: zero = x1      # zero = 0
th2: x2 = zero + 1  # x2 = 1
th2: zero = x2      # zero = 1
th2: x2 = zero - 1  # x2 = 0
th2: zero = x2      # zero = 0
结果：zero = 0

При многопоточности могут возникать такие перекрестные эффекты

初始：zero = 0
th1: x1 = zero + 1  # x1 = 1
th2: x2 = zero + 1  # x2 = 1
th2: zero = x2      # zero = 1
th1: zero = x1      # zero = 1  问题出在这里，两次赋值，本来应该加2变成了加1
th1: x1 = zero - 1  # x1 = 0
th1: zero = x1      # zero = 0
th2: x2 = zero - 1  # x2 = -1
th2: zero = x2      # zero = -1
结果：zero = -1

Когда количество циклов очень велико, неизбежно возникает такой хаос, приводящий к неправильным результатам. Модуль threading предлагает решение: блокировки потоков.

Создайте замок, вchange_zeroФункция должна сначала получить блокировку, чтобы продолжить работу, и, наконец, снять блокировку. Одна и та же блокировка может использоваться только одним потоком за раз, поэтому, когда один поток использует блокировку, другие потоки могут только ждать, пока блокировка будет освобождена, чтобы получить блокировку для вычислений. код изменен на

import threading
import time
zero = 0
lock = threading.Lock()
def change_zero():
    global zero
    for i in range(1000000):
        lock.acquire()
        zero = zero + 1
        zero = zero - 1
        lock.release()
th1 = threading.Thread(target = change_zero)
th2 = threading.Thread(target = change_zero)
th1.start()
th2.start()
th1.join()
th2.join()
print(zero)

Результат, возвращаемый при этом, равен 0 каждый раз

Несколько замечаний:

• acquireа такжеreleaseможно использовать, когдаtry finallyРежим, чтобы гарантировать, что блокировка должна быть снята, иначе могут быть некоторые потоки, ожидающие блокировки, но не могущие ждать
• Хотя использование блокировок потоков может устранить ошибки, вызванные смешиванием переменных, это также снижает эффективность работы, поскольку, когда один поток использует для запуска блокировку, другие потоки не могут выполняться вместе.
• Как правило, не используйтеacquire releaseВключите всю работающую функциональную часть, но только шаг, который может вызвать ошибку, в остальном это ничем не отличается от использования одного потока.
• При использовании нескольких блокировок каждый из двух потоков может удерживать блокировку и пытаться получить блокировку другого, что приведет к взаимоблокировке.Все потоки потребляются здесь и должны быть завершены вручную.
• Когда возникает это явление: например, если для чтения и записи одного и того же файла используется несколько потоков, следует добавить блокировку до и после всего процесса чтения и записи, чтобы гарантировать, что процесс чтения и записи не будет затронут.

Кроме того, блокировка имеет форму управления контекстом, и приведенный выше код можно переписать как

import threading
import time
zero = 0
lock = threading.Lock()
def change_zero():
    global zero
    for i in range(1000000):
        with lock:
            zero = zero + 1
            zero = zero - 1
th1 = threading.Thread(target = change_zero)
th2 = threading.Thread(target = change_zero)
th1.start()
th2.start()
th1.join()
th2.join()
print(zero)

Наконец объясните два общих понятия (из энциклопедии Baidu)

• Синхронизация потоков: Синхронизация здесь относится к работе в заранее определенном порядке, а слово «щипцы» должно относиться к сотрудничеству, помощи и взаимному сотрудничеству. Так называемая синхронизация означает, что при вызове функции вызов не вернется до тех пор, пока не будет получен результат, и другие потоки не могут вызвать этот метод. в отличие от асинхронного.
• Мьютекс: механизм, который не позволяет двум потокам одновременно читать и записывать один и тот же общий ресурс (например, глобальную переменную).

ГИЛ замок

Полное название блокировки GIL — глобальная блокировка интерпретатора. Любой поток Python должен получить блокировку GIL перед его выполнением. Тогда каждый раз, когда выполняется часть кода, интерпретатор автоматически снимает блокировку GIL, и другие потоки могут конкурировать за этот замок.Только получить, чтобы выполнить программу.

Блокировка GIL является причиной того, что многие люди говорят, что многопоточность Python безвкусна, но на самом деле это проблема с интерпретатором CPython (может не быть блокировки GIL, если вы меняете интерпретатор, но большинство программ Python в настоящее время используют интерпретатор CPython). ).

Первое, что нужно сказать, это то, что блокировки GIL влияют только на эффективность программ, интенсивно использующих ЦП. Для программ с интенсивным вводом-выводом или запросов веб-страниц многопоточность по-прежнему очень эффективна, поскольку они в основном потребляют время ожидания.

Для программ с интенсивным использованием ЦП влияние блокировки GIL заключается в том, что при ее существовании многопоточность не может использовать преимущества многоядерности, то есть для выполнения кода может использоваться только одно ядро ​​ЦП. интерпретатор без блокировки GIL). Существование блокировок позволяет одновременно выполнять вычисления только одному потоку, поэтому, даже если многопоточность включена, она не может работать одновременно, а работает линейно. Таким образом, иногда включение многопоточности для запуска программ, интенсивно использующих ЦП, снижает эффективность работы, поскольку больше времени тратится на переключение между потоками и борьбу за блокировки.

тупик

Неправильное использование блокировок может привести к взаимоблокировкам, давайте рассмотрим следующий пример.

import threading
import time
lock1 = threading.Lock()
lock2 = threading.Lock()
class MyThread(threading.Thread):
    def print1(self):
        lock1.acquire() # 获得第一个锁
        print('print1 first ' + threading.current_thread().name)
        time.sleep(1)
        lock2.acquire() # 未释放第一个锁就请求第二个锁
        print('print1 second ' + threading.current_thread().name)
        lock2.release()
        lock1.release()
    def print2(self):
        lock2.acquire() # 获得第二个锁
        print('print2 first ' + threading.current_thread().name)
        time.sleep(1)
        lock1.acquire() # 未释放第二个锁就请求第一个锁
        print('print2 second ' + threading.current_thread().name)
        lock1.release()
        lock2.release()
    def run(self):
        self.print1()
        self.print2()
th1 = MyThread()
th2 = MyThread()
th1.start()
th1.join()
th2.start()
th2.join()
print('finish')

Приведенный выше код заключается в том, что один поток запускается и запускает следующий поток, поэтому проблем с запуском не будет, а результат будет выведен

print1 first Thread-1
print1 second Thread-1
print2 first Thread-1
print2 second Thread-1
print1 first Thread-2
print1 second Thread-2
print2 first Thread-2
print2 second Thread-2
finish

будетstart joinЭта часть изменена на

th1.start()
th2.start()
th1.join()
th2.join()

То есть два потока будут открыты одновременно, и программа выведет

print1 first Thread-1
print1 second Thread-1
print2 first Thread-1
print1 first Thread-2

Он остановится и попадет в состояние взаимоблокировки, и программа никогда не завершится.Фундаментальная причина в том, что один поток держит блокировку 1 и одновременно запрашивает блокировку 2, а другой поток держит блокировку 2 и одновременно запрашивает блокировку 1. , и они не могут получить друг друга. Блокировка не отпускает свою собственную блокировку, и программа вот так заблокирована.

Давайте проанализируем детали

• Первый поток выполняется первымprint1, получает блокировку 1 и ждет 1 секунду. В это время открыт второй поток, пытающийся получить блокировку 1, но не может, так что подождите.
• Первый поток ждет окончания времени, получает блокировку 2 и освобождает две блокировки после печати. Выполнить сразу послеprint2, и получите блокировку 2, подождите 1 секунду
• В это время второй поток может получить блокировку 1 и начать выполнение.print1, также подождите 1 секунду
• По истечении времени ожидания первый поток удерживает блокировку 2 и пытается получить блокировку 1, а первый поток удерживает блокировку 1 и пытается получить блокировку 2, и возникает взаимоблокировка.

(На самом деле, этот пример можно немного упростить, и два потока будут выполняться отдельно.print1 print2может)

Когда мы пишем многопоточные программы, мы должны обращать внимание на то, чтобы избежать взаимоблокировок.

RLock (рекурсивная блокировка, повторная блокировка)

Выше мы инициализируем блокировку, используяthreading.Lock, которая является инструкцией синхронизации потока самого низкого уровня.

Теперь попробуем другойthreading.RLock, что то же самое, чтоLockРазница в том,

• Эта же нить можетRLockзапрашивать несколько раз, покаLockТолько один раз, но при многократном запросеacquireдолжно быть равноreleaseтакое же количество раз
• Lockпо ниткеacquireпосле этого может быть сделано другим потокомrelease,а такжеRLockдолжна быть эта ветка

Давайте посмотрим на следующий пример

import threading
import time
lock = threading.RLock()
def myprint():
    print('start')
    lock.acquire()
    lock.acquire()
    print('try rlock')
    lock.release()
    lock.release()
myprint()

Этот код будет распечатан, как и ожидалось

start
try rlock

если мы используемlock = threading.Lock(), это автоматически создает тупик, потому чтоLockМожно запросить только один раз, поэтому второй раз будет ждать вечно.

RLockкакая функция?

НижеstackoverflowПример сверху

Напишите три функции с вложенными отношениями между ними

def f():
  g()
  h()
def g():
  h()
  do_something1()
def h():
  do_something2()

Теперь хотите заблокировать эти функции, используйтеRLockможно написать так

import threading
lock = threading.RLock()
def f():
  with lock:
    g()
    h()
def g():
  with lock:
    h()
    do_something1()
def h():
  with lock:
    do_something2()

ноLockпросто нужно написать

import threading
lock = threading.Lock()
def f():
  with lock:
    _g()
    _h()
def g():
  with lock:
    _g()
def _g():
  _h()
  do_something1()
def h():
  with lock:
    _h()
def _h():
  do_something2()

потому что с помощьюLockпри звонкеfзаперт, внутриgЕсли его нельзя заблокировать, необходимо определить еще один_gдля представления разблокированной функции. Это преимущество нескольких блокировок, которые могут быть получены одним и тем же потоком.