Яка різниця між процесами та потоками в python?

На даний момент я маю багато-потоковий додаток написаний на py3.7, який працює з апі месенджера. Обробкою кожного повідомлення займається окремий потік. У своєму додатку я використовую sqlite3, для роботи з базою даних, і нещодавно я натрапив на лок бази даних, ймовірно через те, що кожен створений потік так чи інакше взаємодіяв з базою даних, і в якийсь момент потоки перетнулися , створюючи запис у БД одночасно. Команда fuser показала, що ту саму базу даних використовувало відразу кілька потоків одночасно навіть після зупинки головної програми. Звідси питання - яка принципова різниця між потоками та процесами в python, і чи закриваються процеси/потоки автоматично після завершення завдання? Так само якщо я заміню всі свої процеси на потоки - чи залишається ймовірність повторного локу бази даних? Для роботи з БД використовую PeeWee ORM'ку.

2 відповіді 2

Потоки це ті самі потоки, що і в інших мовах, але вони синхронізуються через GIL. Одночасно виконується лише один потік. Змінні потоки загальні.

Процеси виконуються окремо і кожен має свій окремий GIL. Процеси виконуються одночасно. Змінні не вміщені у спільну пам'ять у кожного процесу свої. Процеси можуть завершуватися разом із головним, а можуть завершитися.

Для обчислень із навантаженням на CPU краще використовувати процеси. Для очікування вводу-виводу, роботи з файлами та мережею підійдуть і потоки.

На потоках у вас буде одне підключення до бази даних.

Багатопоточність у Python: модуль threading

Сучасне програмне забезпечення проектується так, що його функції та завдання можуть виконуватись паралельно.Python надає програмісту потужний набір інструментів для роботи з потоками в бібліотеці threading.

Як працює багатопоточність

Багатопотоковість - це виконання програми відразу в декількох потоках, які виконують її функції одночасно.

Багатопотокове програмування можна сплутати з мультипроцесорним. Насправді їхні концепції дуже схожі, але якщо у першому випадку програма працює з потоками, то в іншому – з процесами. Різниця між потоками та процесами проста: потоки мають спільну пам'ять, тому зміни в одному потоці видно в інших, а процеси використовують різні області пам'яті.

Насправді, якщо розглянути одноядерний процесор, операції із різних потоків не виконуються паралельно. Одне ядро ​​може виконати лише одну операцію в одиницю часу, але так як операції виконуються дуже швидко, створюється відчуття паралельного виконання, псевдопаралельність. По-справжньому паралельно програми можуть працювати тільки багатоядерних процесорах, де кожне ядро ​​може виконувати операції незалежно від інших.

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

Однак при виконанні багатопотокової програми на одноядерному процесорі, її продуктивність буде нижчою, ніж якби вона була написана в один потік. Це відбувається тому, що на реалізацію та керування потоками витрачається додаткова пам'ять.

Чи можна вважати threading багатопоточним?

У Python використовується GIL (Global Interpreter Lock), який однопотоковий.Всі потоки, які створюються за допомогою threading будуть працювати всередині потоку GIL. У зв'язку з цим вони будуть оброблятися лише одним ядром. Ні про яку роботу одночасно на кількох фізичних ядрах процесора не може бути й мови.

[su_vnimanie]А оскільки threading буде виконуватися тільки одному ядрі процесора, то немає переваги за швидкістю, тільки навпаки — threading уповільнить роботу.[/su_vnimanie]

Але без нього нікуди не подітися, якщо вам потрібно виконувати кілька завдань одночасно:

• Обробляти натискання кнопки в графічному інтерфейсі, наприклад, за допомогою Tkinter. Якщо після натискання кнопки треба робити багато дій, які вимагають часу, то ці дії треба виконувати в іншому потоці, щоб графічний інтерфейс не підвис на цей час. Відповідно, кнопки треба блокувати, а як потік завершить обчислення — назад розблокувати.
• Якщо наша програма працює одночасно з кількома підключеними пристроями. Вони можуть бути підключені до різних COM-портів.
• Якщо ми завантажуємо файли з мережі та одночасно обробляємо вже завантажені.
• І так далі…

[su_sovet]Якщо нам потрібно, щоб наша програма працювала на кількох фізичних ядрах процесора одночасно, слід звернути увагу на інший модуль — Multiprocessing.[/su_sovet]

У чому переваги тоді модуля Threading в порівнянні з Multiprocessing? Розглянемо їх:

• Простота використання.
• Простіше передавати дані з потоку в основну програму. Взагалі, можна навіть використовувати глобальні змінні. Але при цьому треба правильно проектувати програму, щоб не було помилок, пов'язаних із «Станом перегонів», які ми розглянемо нижче.

[su_sovet]Таким чином, якщо наша програма запускатиметься на одноядерному комп'ютері або навантаження на процесор буде невеликим, то Threading — оптимальний вибір.[/su_sovet]

Підключення бібліотеки threading

Threading – це стандартний модуль, який постачається разом із інтерпретатором. Програмістові не потрібно встановлювати його, достатньо просто підключити модуль за допомогою команди:

Працювати з потоками можна, створюючи екземпляри класу Thread. Щоб створити окремий потік потрібно створити екземпляр класу і застосувати до нього метод start() . Ось приклад:

import threading def myfunc(a, b): print('сума :',a + b) thr1 = threading.Thread(target = myfunc, args = (1, 2)).start() print('основний потік') основний потік сума: 3

Тут ми функцію mydef запустили в окремому потоці. Як аргументи функції передали числа 1 і 2.

threading.Thread()

Ця конструкція дозволяє створити новий потік, створивши екземпляр класу Thread. Ось як виглядають її аргументи:

Вона приймає аргументи:

threading.Thread(group=None, target=None, name=None, args=(), kwargs=<>, *, daemon=None)

Розглянемо їх докладніше:

• group. Має значення None, що зарезервована для майбутнього розширення при реалізації класу ThreadGroup.
• target. Ця функція, яка виконується в потоці за допомогою методу run(), якщо передано значення None, не викликається.
• name. Це ім'я потоку, за замовчуванням воно приймає значення Thread-X, де X - десяткове число. Програміст може вказати ім'я вручну.
• args. Це кортеж, в якому зберігаються аргументи, що передаються в функцію, що викликається.
• kwargs. Це словник, у якому зберігаються аргументи, що передаються у функцію.
• daemon. Це параметр, який встановлює, чи потік є демонічним. За замовчуванням має значення None, тоді властивість daemonic успадковується від потоку. Програміст може самостійно встановити значення параметра.

Демони

Демонами називають процеси, які працюють у фоновому режимі. Python для демона є більш конкретне значення: демонічний потік або потік демона.На відміну від звичайних потоків, потік демона автоматично завершує свою роботу при закритті програми. Іншими словами, програма не чекатиме завершення демонічного потоку, при її закритті ці потоки знищуються, в якому стані вони не знаходилися.

Демонічні потоки використовують для виконання операцій, що виконуються у нескінченному циклі. В інших випадках зазвичай використовують прості потоки, які затримують закриття програми, доки не завершать виконання всіх операцій. Використання демонічних потоків дозволяє операції у фоновому режимі, які зазвичай не пов'язані зі зміною та збереженням довгострокових даних.

Наприклад, якщо програма повністю перезаписує вміст файлу, і механізм перезапису реалізований демонічному потоці, то при несподіваному виході з програми дані втратяться.

У демонічні потоки часто поміщають функції малювання графічного інтерфейсу. Малювання інтерфейсу - нескінченний процес, який завершується відразу після виходу з програми, якщо просто помістити його у звичайний потік, це буде перешкоджати закриттю програми.

Методи роботи з потоками

Для створення та управління потоками використовуються різні методи класу Thread. З їхньою допомогою можна легко маніпулювати відразу кількома потоками і визначати їхню поведінку.

start()

Він використовується для запуску створеного потоку. Після використання threading.Thread() створюється новий потік, але він неактивний. Щоб він почав роботу, використовується метод start().

import threading def myfunc(a, b): print('сума :',a + b) thr1 = threading.Thread(target = myfunc, args = (1, 2)) thr1.start()

Тут поки ми не викликали метод start, функція myfunc не буде запущена.

join()

Цей метод блокує виконання потоку, що його викликав, доки не завершиться потік, метод якого був викликаний.Тобто якщо в потоці thread1 був викликаний метод потоку thread2: thread2.join(), то потік thread1 буде припинено доти, доки виконання thread2 не завершиться.

За допомогою цього методу можна змусити програму дочекатися завершення демонічного потоку. Наприклад, якщо викликати метод в основному потоці, то програма не завершиться, доки не виконається демонічний потік.

Метод join() має аргумент timeout . За замовчуванням він має значення None, але програміст може передати до нього число з точкою, що плаває.

[su_sovet]Якщо аргумент має значення за умовчанням, то виконання потоку припиняється, поки виконується потік методу.[/su_sovet]

Якщо передати як аргумент число, то для методу join() встановиться час очікування, коли воно закінчиться, потік продовжить свою роботу.

Наприклад, thr1.join(100) означає, що буде очікувати завершення виконання потоку thr1 не більше 100 секунд.

Так як метод join() завжди повертає None, щоб перевірити, чи встиг повністю виконатися потік за вказаний timeout, потрібно перевірити, чи потік виконується за допомогою методу is_alive().

import threading import time def myfunc(a, b): time.sleep(2.5) print('сума :', a + b) thr1 = threading.Thread(target = myfunc, args = (1, 2), daemon=True ) thr1.start() thr1.join(0.125) if thr1.is_alive(): print('потік не встиг завершитися') else: print('обчислення завершено') потік не встиг завершитися

Тут ми робимо потік демонічним, щоб програма не чекала закінчення виконання функції. Підключаємо модуль time, щоб зробити затримку в функції на 2.5 секунд. Після старту потоку ми призупиняємо основний потік на 0.125 секунд. Потім виконуємо перевірку is_alive(). Якщо виведе True, то потік не закінчив виконання за 0.125 секунди.

run()

У цьому вся методі описуються операції, виконувані потоком. Він використовується, коли явно створюється екземпляр класу. Приклад:

import threading як th import time class Thr1(th.Thread): # Створюємо екземпляр потоку Thread def __init__(self, var): th.Thread.__init__(self) self.daemon = True # Вказуємо, що цей потік - демон self. var = var # це інтервал, що передається як аргумент def run(self): # метод, який виконується при запуску потоку num = 1 while True: y = num*num + num / (num - 10) # Обчислюємо функцію num += 1 print("При num =", num, " функція y is_alive" >is_alive() 
Метод перевіряє чи виконується потік в даний момент. Його часто використовують у зв'язці з методом join(). Крім того, за його допомогою можна грамотно керувати виконанням потоків демонів, не дозволяючи їм несподівано завершити роботу під час закриття програми, наприклад:
while True: if thr1.is_alive() == True: # Перевіряємо, чи виконується потік демон time.sleep(1) # Якщо так, чекаємо 1 секунду і перевіряємо знову else: break # Якщо ні, виходимо з циклу і закриваємо програму
Зупинка потоку
Бувають ситуації, коли потрібно зупинити потік, що працює у фоні. Допустимо у нас потік у якого в функції run нескінченний цикл. В основній програмі нам потрібно зупинити його.  Тут найпростіше - це створити якусь змінну stop:

У нескінченному циклі робитиме постійно її перевірку і якщо вона True, то завершуватиме його.
Не використовувати функції, які можуть заблокувати виконання на тривалий час. Завжди використовувати timeout.

Ось приклад такої програми:
import threading stop = False def myfunc(): Global stop while stop == False: pass thr1 = threading.Thread(target = myfunc) thr1.start() stop = True while thr1.is_alive() == True: pass print( 'потік завершився')
Тут використовуємо глобальну змінну stop. Коли нам потрібно зупинити потік, ми їй присвоюємо значення True, а далі просто чекаємо на його завершення.
Стан гонки
Стан гонки або race condition – це помилка, що виникає при неправильному проектуванні багатопотокової програми. Вона виникає тоді, коли кілька потоків звертаються до тих самих даних. Наприклад, змінна зберігає число, яке намагаються одночасно змінити потоки thread1 та thread2, що призводить до непередбачуваних результатів або помилок.
Поширена ситуація, коли один потік перевіряє значення змінної на виконання умови, щоб зробити якусь дію, але між перевіркою умови та виконанням дії втручається другий потік, який змінює значення змінної, що призводить до отримання неправильних результатів, наприклад:
x = 5 # Thread 1: if x == 5: # Потік 1 перевіряє умову і вважає його вірним # Thread 2: x = 1 # Потік два змінює значення змінної # Thread 1: print("При x = 5 функція 2*x dostup-k-obshhim-resursam-lock" >Доступ до загальних ресурсів (lock) 
Для того щоб запобігти стану гонки, потрібно використовувати блокування threading.Lock(), яка не дозволяє відразу кільком потокам працювати з одними і тими ж даними. Іншими словами, Lock захищає дані від одночасного доступу.
threading.Lock() – повертає об'єкт, який, образно висловлюючись, є дверима в кімнату, яка замикається, якщо в кімнаті хтось знаходиться. Тобто якщо потік використовував Lock (увійшов до кімнати), то інший потік змушений чекати до тих пір, поки потік, що використовував Lock, не відмовиться від нього (вийде з кімнати).
Отриманий об'єкт має два методи: acquire() і release(). Розглянемо їх.
acquire()
Метод дозволяє потоку отримати блокування. Має два аргументи: blocking та timeout.
Коли викликається з аргументом blocking рівним True (значення за замовчуванням), блокує Lock, доки він не буде розблокований і повертає True.Якщо об'єкт вже заблокований, потік зупиняється і чекає, доки об'єкт не буде розблокований, а потім сам блокує його.
При виклику з аргументів False, якщо об'єкт Lock розблоковано, метод блокує його та повертає True. Якщо Lock вже заблоковано, метод нічого не робить та повертає False.
Аргумент timeout (за замовчуванням -1) можна змінити тільки якщо аргумент блокування має значення True. Якщо аргумент передати позитивне значення, то об'єкт блокується на вказану кількість секунд з урахуванням часу очікування блокування. Аргумент за замовчуванням вказує метод використовувати нескінченне очікування.
release()
Цей метод розблокує об'єкт Lock. Інтерпретатор дозволяє викликати його з будь-якого потоку, а не тільки потоку, який заблокував Lock в даний момент.
Метод нічого не повертає та викликає помилку RuntimeError, якщо викликається, коли об'єкт Lock вже розблокований.
Ось приклад:
import threading lock = threading.Lock() x = 'Рython 2' # . lock.acquire() x = 'Rython 3' print(x) lock.release() Python 3
Тут ми створюємо об'єкт lock, за його допомогою ми будемо безпечно зчитувати та змінювати дані. Як дані, які ми блокуватимемо в даному прикладі це одна змінна x. Далі показано безпечну зміну даних: спочатку за допомогою acquire чекаємо своєї черги доступу до них. Потім змінюємо їх (у нашому прикладі перезаписуємо значення змінної з Python 2 на Python 3). Далі виводимо значення консоль. Після цього звільняємо доступ для інших потоків. Якщо всі потоки, яким потрібен буде доступ до даних x використовувати lock, можна уникнути «Стану гонки».
deadlock
При використанні Lock виникає серйозна проблема, яка призводить до повної зупинки програми.Якщо викликати метод acquire(), а об'єкт Lock вже заблокований, то потік, що викликав acquire(), буде чекати, поки заблокований об'єкт потік не викличе release().
Якщо один потік викликає метод блокування кілька разів поспіль, виконання потоку припиняється, поки він сам не викличе release(). Однак він не може викликати release, тому що його виконання призупинено, що означає нескінченне блокування програми.
Можна запобігти самоблокуванню, якщо видалити зайвий виклик acquire(), але це не завжди можливо.  Самоблокування може відбуватися через наступні речі:

Виникнення помилок, коли Lock залишається заблокованим.
Неправильне проектування програми, коли одна функція викликається іншою функцією, яка не має блокування.

У разі виникнення помилок достатньо скористатися конструкцією try-finally або оператором with.
Ось приклад з with:
lock = threading.Lock() with lock: # оператори pass
Конструкція  try-finally дозволяє видаляти блокування навіть у разі виникнення помилок, що дозволяє уникнути deadblock. Приклад:
lock = threading.Lock() lock.acquire() try: # оператори pass finally: lock.release()
Конструкція try-finally гарантує, що код finally буде виконаний завжди, незалежно від помилок і результатів блоку try.
Однак це не працює у разі самоблокування через неправильне проектування програми. Для цього було створено об'єкт RLock.
RLock
Якщо Lock заблокований, він блокує будь-який потік, який спробував зробити те саме, навіть якщо цей потік і є власником блокування в даний момент. Наприклад, програміст написав код:
import threading lock1 = threading.Lock() def part1(): lock1.acquire() try: # обчислити суму елементів першої частини об'єкта pass finally: lock1.release() return sum def part2(): lock1.acquire() try: # Обчислити суму елементів другої частини об'єкта pass finally: lock1.release() return sum def both_parts(): p1 = part1() p2 = part2() return p1, p2
Цей код буде працювати, але його проблема полягає в тому, що при виклику функції both_parts, в ній викликаються функції part1 і part2. Між викликами цих функцій може отримати доступ до даних якийсь інший потік та їх поміняти. А що робити, якщо потрібно уникнути зміни іншим потоком?
Щоб вирішити проблему, потрібно заблокувати lock1 і both_parts , перепишемо її:
def both_parts(): lock1.acquire() try: p1 = part1() p2 = part2() finally: lock1.release() return p1, p2
Ідея проста: зовнішня both_parts блокує потік на час виконання функцій part1 та part1. Кожна з функцій блокує потік для підсумовування своєї частини об'єкта. Однак об'єкт Lock не дозволить цього зробити, цей код призведе до повного зависання програми, тому що для Lock немає різниці, де в потоці викликали acquire().
RLock блокує потік, лише якщо об'єкт заблокований іншим потоком. Використовуючи RLock, потік ніколи не зможе заблокувати себе.
Використовувати RLock потрібно для керування вкладеним доступом до об'єктів, що розділяються. Щоб вирішити проблему з Lock в коді вище, достатньо замінити рядок «lock1 = threading.Lock() » на «lock1 = threading.RLock() ».
[su_vnimanie]Також слід пам'ятати, що, хоча можна викликати acquire() кілька разів, метод release() потрібно викликати стільки ж раз. При кожному виклик acquire() рівень рекурсії збільшується на одиницю, відповідно при кожному виклик release() він зменшується на одиницю.[/su_vnimanie]
Передача даних за допомогою черг (Queue)
Для передачі даних за допомогою черг використовується клас Queue з бібліотеки queue, який імпортується командною: from queue import Queue.
Бібліотека queue містить усі необхідні інструменти передачі даних між потоками і реалізує потрібні механізми блокування.
Клас Queue реалізує чергу FIFO, який працює так: перший елемент, який пішов у чергу, першим і вийде з неї. Цю чергу можна порівняти з вертикальною порожнистою трубою, яку зверху кидають елементи.
Queue має параметр maxsize, що приймає лише цілі чисельні значення. Він вказує максимальну кількість елементів, які можна помістити в чергу. Коли максимум досягається, додавання до черги елементів блокується, поки в ній не звільниться місце. Якщо maxsize набуває значення  
Для взаємодії з чергами використовується Event, об'єкт модуля threading. З його допомогою потік може виконати потрібні операції, коли отримає сигнал від іншого потоку. Крім того, потік не обов'язково повинен припиняти свою роботу на час очікування сигналу.
Для передачі даних та роботи з чергами використовуються методи (працюють з усіма видами черг, а не тільки з Queue):
qsize()
Повертає зразковий розмір черги. Важливо розуміти дві речі:

Якщо qsize() більше за нуль, наступний метод get() все одно може бути заблокований.
Якщо qsize() менший за maxsize, наступний метод put() може бути заблокований.

Це може виникнути через те, що до черги можуть звернутися інші потоки та отримати/записати дані відразу після того, як ви отримали її розмір.
empty()
Метод перевіряє, чи міститься щось у черзі. Якщо черга порожня, повертається True, якщо черга містить елементи, що повертається False.
Як і з методом qsize(), повернення True або False не гарантує, що наступний метод put() чи get() не буде заблоковано.
full()
Перевіряє, чи заповнена чергаЯкщо черга заповнена, повертає True, інакше повертає False.
Як і в попередніх методах повернення True або False не дає гарантій, що put() і get() не будуть заблоковані.
put()
Метод містить новий об'єкт у чергу, має обов'язковий аргумент item і два необов'язкові аргументи: block = True і timeout = None.
Queue.put(item, block=True, timeout=None)
Залежно від зазначених аргументів, очікування місця у черзі поводитиметься по-різному:

Якщо аргумент block має значення True, а timeout — None, об'єкт, який потрібно завантажити в чергу, нескінченно чекатиме вільного місця.
При timeout більше нуля, очікування вільного місця триватиме не довше за вказану кількість секунд, якщо за цей час вільного місця так і не з'явилося, збудеться виняток.
Якщо блок має значення False, аргумент timeout ігнорується, і елемент можна помістити в чергу, тільки якщо є вільне місце, інакше відразу ж збуджується виняток.

Ось приклад створення черги на Python та додавання до неї елемента:
from queue import Queue queue1 = Queue() x = 'some data' queue1.put(x)
put_nowait()
Еквівалентно виклику put(item, False). поміщає елемент у чергу, тільки якщо є місце, інакше викликає виняток.
get()
Видаляє та повертає елемент з черги.
Queue.get(block=True, timeout=None)
Залежно від значень аргументів очікування об'єкта поводиться по-різному:

Якщо аргументи мають значення за умовчанням, метод очікує об'єкт із черги до того часу, поки той стане доступний.
При timeout - позитивне число, то об'єкт з черги очікується певний час, після якого викликається виняток.
Якщо блок має значення False, елемент повертається, тільки якщо він доступний, інакше викликається виняток (аргумент timeout ігнорується).

Ось приклад. Тут ми додаємо рядок у чергу.
from queue import Queue queue1 = Queue() queue1.put('Python 3') value = queue1.get() print(value) Python 3
get_nowait()
Еквівалентно виклику get(False).
task_done()
Цей метод працює у зв'язку з методом join().
Метод показує, що раніше поставлене завдання було виконано. Після отримання кожного елемента з черги, допустимо за допомогою get(), потрібно викликати task_done(), щоб зменшити лічильник завдань. Нижче описано метод join з прикладом
Якщо task_done() викликається більше разів, ніж кількість елементів, розміщених у чергу, то збуджується виключення ValueError.
join()
Блокує потік, доки всі елементи черги не будуть отримані та оброблені.
Щоразу, коли до черги додається новий елемент, збільшується лічильник незавершених завдань. При виклику task_done(), лічильник зменшується, показуючи, що обробка елемента у черзі завершена і можна переходити до наступного. Коли лічильник дорівнює нулю, з потоку знімається блокування.
from queue import Queue queue1 = Queue() queue1.put('Python 2') queue1.put('Python 3') print(queue1.get()) queue1.task_done() print(queue1.get()) queue1. task_done() queue1.join()
Цей приклад для того, щоб показати роботу join та task_done. Тут все відбувається в одному потоці. Зазвичай пише в чергу один потік дані, потім чекає, коли їх оброблять за допомогою join. Інший потік при отриманні кожного нового значення викликає task_done.
Приклад програми
Суть програми проста: студенти мають здати залікову роботу. Робота здається двом викладачам, час, за який перевірять роботу, залежить від рейтингу студента (що більший рейтинг, тим краща робота, тим менше перевіряють).
з import import Queue import time, datetime, threading students= [(99, "Андрій"), (76, "Олександр"), (75, "Микита"), (72, "Євген"), (66, "Олексій "), (62, "Сергій"), (50, "Михайло")] def student(q): while True: # Отримуємо завдання з черги check = q.get() # Виводимо час початку перевірки print(check[1], 'здав роботу в', datetime.datetime.now() .strftime('%H:%M:% S')) #Час витрачений на перевірку, який залежить від рейтингу time.sleep((100-check[0])/5) # Час закінчення перевірки print(check[1], 'забрав роботу в', datetime.datetime.now() .strftime('%H:%M:%S')) # Даємо сигнал про те, що завдання черги виконано q .task_done() # Створюємо чергу q = Queue() # Завантажуємо в чергу студентів for x in students: q.put(x) #створюємо і запускаємо потоки thread1 = threading.Thread(target=student, args=(q,), daemon=True) thread2 = threading.Thread(target=student, args=(q,), daemon=True) thread1.start () time.sleep(10) thread2.start() # Блокуємо виконання до завершення всіх завдань q.join() print("Цей текст надрукується після закінчення блокування")
Андрій здав роботу о 18:58:35 Андрій забрав роботу о 18:58:36 Олександр здав роботу о 18:58:36 Олександр забрав роботу о 18:58:40 Микита здав роботу о 18:58:40 Євген здав роботу о 18 :58:45 Микита забрав роботу о 18:58:45 Олексій здав роботу у 18:58:45 Євген забрав роботу о 18:58:51 Сергій здав роботу о 18:58:51 Олексій забрав роботу о 18:58:52 Михайло здав роботу о 18:58:52 Сергій забрав роботу о 18:58:59 Михайло забрав роботу о 18:59:02. Цей текст надрукується після закінчення блокування
Як видно з результатів, студенти оброблялися у два потоки, двома викладачами. q.join() заблокувало виконання основного потоку, тому текст роздрукувався тільки після завершення всіх завдань черги.
Корисні інструменти модуля threading
Threading має ще кілька корисних інструментів, які можуть стати в нагоді для вирішення більш спеціалізованих завдань.
Semaphore
Це один із найстаріших примітивів для синхронізації в історії інформатики. Семафор використовує внутрішній лічильник, який зменшується при кожному виклик acquire() і збільшується при кожному виклик release(). Лічильник не може стати меншим за нуль, коли він стає рівним нулем, acquire() блокує потік.
import threading x = 'Python' sem = threading.Semaphore() sem.acquire() x = 'Python 2' sem.release() with sem: x = 'Python 3'
Тут навів два варіанти отримання доступу до даних:
Timer
Цей клас дозволяє контролювати час запуску будь-якої дії. Timer є підклас Thread.
Ось його аргументи:
threading.Timer(interval, function, args=None, kwargs=None)
Таймери запускаються так само, як і потоки, за допомогою методу start(). Їх можна зупинити, використовуючи метод cancel().
За допомогою таймера програміст може викликати функцію, присвоїти значення змінної або, наприклад, запустити потік у певний час.
Приклад використання:
import threading def myfunc(): print('tick-tack') timer = threading.Timer(4, myfunc) timer.start() tick-tack
Тут функція myfunc виконається через 4 секунди після виклику методу start().
Barrier
Цей клас дозволяє реалізувати простий механізм синхронізації потоків. Його можна використовувати для фіксованого числа потоків, коли необхідно, щоб кожен потік чекав на виконання будь-якої дії всіма.
Для того, щоб продовжити виконання, всі потоки повинні викликати метод wait(), якщо хоч один потік не зробив цього, інші блокуються доти, доки метод не буде викликаний.
Так виглядають його аргументи:
threading.Barrier(parties, action=None, timeout=None)
Розглянемо приклад використання:
import threading import time barrier = threading.Barrier(2) def myfunc(): barrier.wait() print('відпрацював barrier') thr1 = threading.Thread(target = myfunc).start() time.sleep(1) print ('основний потік') barrier.wait() основний потік відпрацював barrier
Тут виставляю barrier на 2 виклики wait.Тобто для того, щоб виконався код після wait, wait повинен бути викликаний у 2 потоках. В даному випадку функція myfunc відразу запускається в потоці, але вона відразу не виведе 'відпрацював barrier' в консоль, а дочекається коли в основному потоці буде викликаний wait теж.
Event
Event є простим механізмом реалізації зв'язку між потоками: один потік дає сигнал про подію, інші очікують цього сигналу.
Об'єкт події управляє внутрішнім прапором, який може бути встановлений у True або False за допомогою методів set() та clear(). Також є методи is_set(), якими можна перевірити стан внутрішнього прапора. За допомогою методу wait(timeout=None) можна чекати, поки не виставлений прапор у True. Також за необхідності можна задати час очікування.
Ось приклад:
import threading import time event = threading.Event() def myfunc(): time.sleep(1) event.set() thr1 = threading.Thread(target = myfunc).start() print(event.is_set()) event .wait() print(event.is_set()) event.clear() print(event.is_set()) False True False
Висновок
Можливість керування потоками в Python – це потужний інструмент розробки великих програм. Для роботи з ними використовується модуль Threading та бібліотека queue у зв'язці з ним.
Кожен програміст Python повинен вміти працювати з потоками, чергами та розуміти, як влаштовано блокування, доступ до даних та їх передачу між потоками.