Високопродуктивний Python: Asyncio

Програмування з паралелізмом — це підхід до програмування, який займається одночасним виконанням кількох завдань. У Python asyncio — потужний інструмент для реалізації асинхронного програмування. Заснований на концепції корутин, asyncio може ефективно обробляти завдання, що вимагають інтенсивної роботи з вводу/виводу (I/O). Ця стаття познайомить вас з основними принципами та використанням asyncio.

Чому нам потрібен asyncio

Ми знаємо, що при обробці операцій вводу/виводу використання багатозадачності може значно підвищити ефективність порівняно з роботою на одному потоці.
Отже, чому нам все ж таки потрібен asyncio?

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

• Наприклад, процес виконання багатозадачності легко переривається, тому можуть виникати ситуації гонки (race condition).
• Крім того, існує певна вартість перемикання між потоками, і кількість потоків не можна збільшувати безкінечно. Тому, якщо ваші операції вводу/виводу дуже важкі, багатозадачність навряд чи зможе забезпечити високу ефективність і якість.

Саме для вирішення цих проблем і з'явився asyncio.

Sync VS Async

Давайте спочатку розрізнимо поняття Sync (синхронний) та Async (асинхронний).

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

Як працює asyncio

1. Корутин (Coroutines): asyncio використовує корутини для досягнення асинхронних операцій. Корутина — це спеціальна функція, визначена за допомогою ключового слова async. У корутині можна використовувати ключове слово await, щоб призупинити виконання поточної корутини і чекати на завершення асинхронної операції.
2. Цикл подій (Event Loop): Цикл подій — це один з основних механізмів asyncio. Він відповідає за планування та виконання корутин, а також за перемикання між ними. Цикл подій постійно перевіряє наявність виконуваних завдань. Як тільки завдання готове (наприклад, коли операція вводу/виводу завершена або спливає таймер), цикл подій ставить його в чергу виконання і переходить до наступного завдання.
3. Асинхронні завдання (Async Tasks): У asyncio ми виконуємо корутини, створюючи асинхронні завдання.
   Асинхронні завдання створюються за допомогою функції asyncio.create_task(), яка інкапсулює корутину в об'єкт, що може бути чекаємим (awaitable object), і передає його в цикл подій для обробки.
4. Асинхронні операції вводу/виводу (I/O Operations): asyncio надає набір асинхронних операцій вводу/виводу (наприклад, мережеві запити, читання та запис файлів тощо), які можуть бути безшовно інтегровані з корутинами та циклом подій через ключове слово await. Використовуючи асинхронні операції вводу/виводу, можна уникнути блокування під час очікування завершення операцій вводу/виводу, що покращує продуктивність програми та її конкурентність.
5. Зворотні виклики (Callbacks): asyncio також підтримує використання функцій зворотного виклику для обробки результатів асинхронних операцій. Функцію asyncio.ensure_future() можна використовувати для інкапсуляції функції зворотного виклику в об'єкт, що може бути чекаємим, і передавання її в цикл подій для обробки.
6. Конкурентне виконання (Concurrent Execution): asyncio може одночасно виконувати кілька корутинних завдань.
   Цикл подій автоматично плануватиме виконання корутин згідно з готовністю завдань, забезпечуючи таким чином ефективне конкурентне програмування.

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

Корутини та асинхронне програмування

Корутини — важливе поняття в asyncio. Це легковажні одиниці виконання, які можуть швидко перемикатися між завданнями без витрат, що виникають через перемикання потоків.
Корутини можна визначати за допомогою ключового слова async, а ключове слово await використовується для того, щоб призупинити виконання корутини та відновити його після завершення певної операції.

Ось простий приклад коду, що демонструє, як використовувати корутини для асинхронного програмування:

import asyncio  

async def hello():  
 print("Hello")  
 await asyncio.sleep(1) # Імітація операції, що займає час  
 print("World")  
# Створення циклу подій  
loop = asyncio.get_event_loop()  
# Додавання корутини до циклу подій та її виконання  
loop.run_until_complete(hello())

У цьому прикладі функція hello() є корутиною, визначеною за допомогою ключового слова async. Усередині корутини ми можемо використовувати await, щоб призупинити її виконання. Тут asyncio.sleep(1) використовується для імітації операції, що займає час.
Метод run_until_complete() додає корутину до циклу подій та виконує її.

Асинхронні операції вводу/виводу

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

Ось простий приклад коду, що демонструє, як використовувати asyncio для асинхронних мережевих запитів:

import asyncio  
import aiohttp  

async def fetch(session, url):  
 async with session.get(url) as response:  
 return await response.text()  
async def main():  
 async with aiohttp.ClientSession() as session:  
 html = await fetch(session, 'https://www.example.com')  
 print(html)  
# Створення циклу подій  
loop = asyncio.get_event_loop()  
# Додавання корутини до циклу подій та її виконання  
loop.run_until_complete(main())

У цьому прикладі ми використовуємо бібліотеку aiohttp для мережевих запитів.
Функція fetch() є корутиною. Вона ініціює асинхронний GET-запит через метод session.get() і чекає на відповідь, використовуючи ключове слово await. Функція main() є ще однією корутиною. Вона створює об'єкт ClientSession для повторного використання, а потім викликає метод fetch(), щоб отримати вміст веб-сторінки та вивести його.

Примітка: Тут ми використовуємо aiohttp замість бібліотеки requests, оскільки бібліотека requests не сумісна з asyncio, тоді як aiohttp сумісна. Для ефективного використання asyncio, зокрема для використання його потужних функцій, в багатьох випадках необхідні відповідні бібліотеки Python.

Паралельне виконання кількох задач

asyncio також надає деякі механізми для паралельного виконання кількох задач, такі як asyncio.gather() та asyncio.wait().
Ось приклад коду, який показує, як використовувати ці механізми для паралельного виконання кількох корутин:

import asyncio  

async def task1():  
 print("Завдання 1 розпочато")  
 await asyncio.sleep(1)  
 print("Завдання 1 завершено")  
async def task2():  
 print("Завдання 2 розпочато")  
 await asyncio.sleep(2)  
 print("Завдання 2 завершено")  
async def main():  
 await asyncio.gather(task1(), task2())  
# Створити цикл подій  
loop = asyncio.get_event_loop()  
# Додати корутину в цикл подій і виконати  
loop.run_until_complete(main())

У цьому прикладі ми визначаємо дві корутини task1() і task2(), які обидві виконують деякі ресурсоємні операції. Корутина main() запускає ці два завдання одночасно через asyncio.gather() і чекає їх завершення.
Паралельне виконання може покращити ефективність виконання програми.

Як вибрати?

У реальних проектах, чи слід нам вибирати багатозадачність (multithreading) чи asyncio? Один досвідчений фахівець яскраво підсумував це так:

if io_bound:  
 if io_slow:  
 print('Використовувати Asyncio')  
 else:  
 print('Використовувати багатозадачність')  
elif cpu_bound:  
 print('Використовувати багатопроцесорність')

• Якщо операції з вводу/виводу (I/O) є основними і вони повільні, вимагаючи співпраці багатьох завдань/потоків, то використання asyncio буде більш доцільним.
• Якщо операції з вводу/виводу (I/O) швидкі і потрібна лише обмежена кількість завдань/потоків, то достатньо буде багатозадачності.
• Якщо обмеження на виконання програми пов'язане з процесором (CPU bound), то необхідно використовувати багатопроцесорність для покращення ефективності виконання програми.

Практика

Введіть список.
Для кожного елемента в списку ми хочемо обчислити суму квадратів всіх цілих чисел від 0 до цього елемента.

Синхронна реалізація

import time  

def cpu_bound(number):  
 return sum(i * i for i in range(number))  
def calculate_sums(numbers):  
 for number in numbers:  
 cpu_bound(number)  
def main():  
 start_time = time.perf_counter()  
 numbers = [10000000 + x for x in range(20)]  
 calculate_sums(numbers)  
 end_time = time.perf_counter()  
 print('Обчислення займає {} секунд'.format(end_time - start_time))  
if __name__ == '__main__':  
 main()

Час виконання: Обчислення займає 16.00943413000002 секунд

Асинхронна реалізація з використанням concurrent.futures

import time  
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed  

def cpu_bound(number):  
 return sum(i * i for i in range(number))  
def calculate_sums(numbers):  
 with ProcessPoolExecutor() as executor:  
 results = executor.map(cpu_bound, numbers)  
 results = [result for result in results]  
 print(results)  
def main():  
 start_time = time.perf_counter()  
 numbers = [10000000 + x for x in range(20)]  
 calculate_sums(numbers)  
 end_time = time.perf_counter()  
 print('Обчислення займає {} секунд'.format(end_time - start_time))  
if __name__ == '__main__':  
 main()

Час виконання: Обчислення займає 7.314132894999999 секунд

У цьому покращеному коді ми використовуємо concurrent.futures.ProcessPoolExecutor для створення пулу процесів, а потім використовуємо метод executor.map() для подачі завдань і отримання результатів.
Зверніть увагу, що після використання executor.map(), якщо вам потрібно отримати результати, можна пройтися по результатах і перетворити їх на список або використати інші методи для обробки результатів.

Реалізація з багатозадачністю

import time  
import multiprocessing  

def cpu_bound(number):  
 return sum(i * i for i in range(number))  
def calculate_sums(numbers):  
 with multiprocessing.Pool() as pool:  
 pool.map(cpu_bound, numbers)  
def main():  
 start_time = time.perf_counter()  
 numbers = [10000000 + x for x in range(20)]  
 calculate_sums(numbers)  
 end_time = time.perf_counter()  
 print('Обчислення займає {} секунд'.format(end_time - start_time))  
if __name__ == '__main__':  
 main()

Час виконання: Обчислення займає 5.024221667 секунд

concurrent.futures.ProcessPoolExecutor та multiprocessing — це обидві бібліотеки для реалізації багатопроцесорної паралельності в Python. Існують деякі відмінності:

1.
Інкапсуляція на основі інтерфейсу: concurrent.futures.ProcessPoolExecutor — це високорівневий інтерфейс, наданий модулем concurrent.futures. Він інкапсулює функції багатопроцесорної обробки, що робить написання багатопроцесорного коду простішим. У той час як multiprocessing є однією з стандартних бібліотек Python, що забезпечує повну підтримку багатопроцесорної обробки та дозволяє безпосередньо працювати з процесами.

1. Використання API: Використання concurrent.futures.ProcessPoolExecutor схоже на використання пулу потоків. Він передає викликаються об'єкти (такі як функції) в пул процесів для виконання та повертає об'єкт Future, який можна використовувати для отримання результату виконання. multiprocessing надає більше низькорівневих інтерфейсів для управління процесами та комунікації. Процеси можна явно створювати, запускати та контролювати, а комунікація між кількома процесами здійснюється за допомогою черг або труб.
   3.
   Масштабованість та гнучкість: Оскільки multiprocessing надає більше низькорівневих інтерфейсів, він є більш гнучким у порівнянні з concurrent.futures.ProcessPoolExecutor. Безпосередньо працюючи з процесами, можна досягти більш детального контролю за кожним процесом, наприклад, встановлювати пріоритети процесів і обмінюватися даними між процесами. concurrent.futures.ProcessPoolExecutor більше підходить для простого паралелізму завдань, приховуючи багато внутрішніх деталей і полегшуючи написання багатопроцесорного коду.

2. Кросплатформна підтримка: Як concurrent.futures.ProcessPoolExecutor, так і multiprocessing забезпечують кросплатформну підтримку багатопроцесорності і можуть використовуватися на різних операційних системах.

Підсумовуючи, concurrent.futures.ProcessPoolExecutor є високорівневим інтерфейсом, що інкапсулює функції багатопроцесорної обробки, підходить для простого паралелізму багатопроцесних завдань.
`multiprocessing є більш низькорівневою бібліотекою, яка надає більше контролю та гнучкості, підходить для сценаріїв, що потребують детального контролю за процесами. Потрібно обирати відповідну бібліотеку залежно від конкретних вимог. Якщо необхідний лише простий паралелізм завдань, можна використатиconcurrent.futures.ProcessPoolExecutor, щоб спростити код; якщо потрібен більш низькорівневий контроль та комунікація, варто використовувати бібліотекуmultiprocessing`.

Підсумки

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

Завдання в asyncio не перериваються під час виконання, тому ситуація з умовою гонки (race condition) не виникає.

Особливо в сценаріях з інтенсивними операціями вводу/виводу (I/O), asyncio має вищу ефективність роботи, ніж багатопоточність.
`Оскільки вартість перемикання завдань уasyncioзначно менша, ніж у багатопоточності, і кількість завдань, які може запускатиasyncio`, набагато більша за кількість потоків у багатопоточних програмах.

Проте варто зазначити, що в багатьох випадках використання asyncio потребує підтримки специфічних сторонніх бібліотек, таких як aiohttp у попередньому прикладі.
``Якщо операції введення/виведення (I/O) швидкі та не надто ресурсозатратні, використання багатопоточності також може ефективно вирішити проблему.

• asyncio — це бібліотека Python для реалізації асинхронного програмування.
• Корутини (coroutines) є основною концепцією asyncio, що дозволяє виконувати асинхронні операції за допомогою ключових слів async та await.
• asyncio надає потужний API для асинхронних операцій введення/виведення та легко справляється з задачами, що потребують інтенсивного використання I/O.
• За допомогою таких механізмів, як asyncio.gather(), кілька корутин можна виконувати паралельно.

Leapcell: Ідеальна платформа для FastAPI, Flask та інших Python-застосунків

Наприкінці хочу представити ідеальну платформу для розгортання Flask/FastAPI: Leapcell.

Leapcell — це хмарна платформа, спеціально розроблена для сучасних розподілених застосунків.
``Модель ціноутворення "плати за використання" гарантує відсутність витрат на простій, тобто користувачі сплачують лише за ресурси, які вони реально використовують.

1. Підтримка кількох мов програмування

• Підтримується розробка на JavaScript, Python, Go або Rust.

1. Безкоштовне розгортання необмеженої кількості проєктів

• Стягується плата тільки за використання. Немає платіжних вимог, коли немає запитів.

1. Неперевершена ефективність витрат

• Плата за використання, без зборів за простій.
• Наприклад, $25 може обслуговувати 6,94 мільйона запитів з середнім часом відгуку 60 мілісекунд.

1. Спрощений досвід розробника

• Інтуїтивно зрозумілий інтерфейс для легкого налаштування.
• Повністю автоматизовані CI/CD пайплайни та інтеграція GitOps.
• Реальні метрики та логи, що надають корисну інформацію для дій.

1.
``Безпроблемна масштабованість і висока продуктивність

• Автоматичне масштабування для обробки високої кількості одночасних запитів без проблем.
• Відсутність накладних витрат на операції, що дозволяє розробникам зосередитись на розробці.

Дізнайтеся більше в документації! Twitter Leapcell: https://x.com/LeapcellHQ

Перекладено з: High-Performance Python: Asyncio