Кратко про библиотеку gevent в Python: высокопроизводительная асинхронная библиотека

Привет, Хабр!

Gevent — это высокопроизводительная асинхронная библиотека для Python, которая решает проблемы многозадачности с помощью корутин, известных как «зеленые потоки» или Greenlets. Зеленые потоки — легковесные корутины, которые позволяют выполнять задачи одновременно без затрат ресурсов, связанных с традиционными потоками.

Также Gevent преуспевает в неблокирующих операциях ввода/вывода и имеется встроенная техника под названием «monkey patching«, которая модифицирует некоторые модули стандартной библиотеки, делая их кооперативными. Так можно преобразовывать блокирующие вызовы в неблокирующие в рамках среды Gevent.

Рассмотрим библиотеку подробней. Но для начала — установим:

pip install gevent

Основной синтаксис

Создание и управление зелеными потоками

gevent.spawn(): создание зеленых потоков

import gevent  def task(message):     print(message)  # создание зеленого потока greenlet = gevent.spawn(task, "Hello, Habr!") # ожидание завершения зеленого потока greenlet.join()

Фнкция создает новый зеленый поток, который выполняет указанную функцию task с переданным аргументом "Hello, Habr!". join() используется для ожидания завершения зеленого потока.

gevent.spawn_later(): отложенный запуск зеленых потоков

import gevent  def delayed_task():     print("Task executed after delay")  # отложенный запуск через 5 секунд greenlet = gevent.spawn_later(5, delayed_task) # ожидание завершения зеленого потока greenlet.join()

Функция запускает зеленый поток через заданное время (в нашем случае 5 секунд).

gevent.kill() и gevent.killall(): остановка зеленых потоков

import gevent  def task():     gevent.sleep(10)     print("This will not be printed")  # создание и запуск зеленого потока greenlet = gevent.spawn(task) # остановка зеленого потока greenlet.kill()  # создание и запуск нескольких зеленых потоков greenlets = [gevent.spawn(task) for _ in range(3)] # остановка всех зеленых потоков gevent.killall(greenlets)

kill() используется для остановки одного зеленого потока, а killall() — для остановки всех переданных зеленых потоков.

gevent.joinall(): ожидание завершения всех зеленых потоков

import gevent  def task(id):     gevent.sleep(id)     print(f"Task {id} completed")  # создание и запуск нескольких зеленых потоков greenlets = [gevent.spawn(task, i) for i in range(5)] # ожидание завершения всех зеленых потоков gevent.joinall(greenlets)

joinall() позволяет ожидать завершения всех зеленых потоков из переданного списка.

Работа с синхронизацией

gevent.lock(): замки

import gevent from gevent.lock import Semaphore  sem = Semaphore(1)  def task1():     with sem:         print("Task 1 acquired lock")         gevent.sleep(2)         print("Task 1 released lock")  def task2():     with sem:         print("Task 2 acquired lock")         gevent.sleep(1)         print("Task 2 released lock")  gevent.joinall([     gevent.spawn(task1),     gevent.spawn(task2) ])

Семафоры используются для управления доступом к ресурсам между зелеными потоками.

gevent.event(): события

import gevent from gevent.event import Event  evt = Event()  def setter():     gevent.sleep(3)     print("Event set")     evt.set()  def waiter():     print("Waiting for event")     evt.wait()     print("Event received")  gevent.joinall([     gevent.spawn(setter),     gevent.spawn(waiter) ])

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

Взаимодействие с I/O

gevent.socket: кооперативные сокеты

import gevent from gevent import socket  def fetch(host):     sock = socket.create_connection((host, 80))     sock.sendall(b"GET / HTTP/1.1\r\nHost: " + host.encode('ascii') + b"\r\n\r\n")     data = sock.recv(1024)     print(f"{host}: {data}")  hosts = ['www.google.com', 'otus.ru', 'habr.com'] jobs = [gevent.spawn(fetch, host) for host in hosts] gevent.joinall(jobs)

С кооперативными сокетами можно выполнять сетевые операции асинхронно, не блокируя выполнение программы.

gevent.subprocess: работа с subprocess в асинхронном режиме

import gevent from gevent import subprocess  def run_command():     process = subprocess.Popen(['ls', '-l'], stdout=subprocess.PIPE)     out, err = process.communicate()     print(out.decode('utf-8'))  gevent.spawn(run_command).join()

С модулем gevent.subprocess можно асинхронно запускать и управлять внешними процессами.

Monkey Patching

gevent.monkey.patch_all(): кооперативный патчинг стандартных библиотек

from gevent import monkey monkey.patch_all()  import socket import requests  def fetch(url):     response = requests.get(url)     print(f"Fetched {len(response.content)} bytes from {url}")  urls = ['http://www.google.com', 'http://www.example.com', 'http://www.python.org'] jobs = [gevent.spawn(fetch, url) for url in urls] gevent.joinall(jobs)

Функция monkey.patch_all() заменяет стандартные блокирующие вызовы на кооперативные.

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

В большинстве примеров будем использовать Flask для создания API сервиса обмена сообщениями.

Веб-сервер с использованием Gevent

Создадим веб-сервер, который может одновременно обслуживать множество клиентов:

from gevent import monkey monkey.patch_all()  from gevent.pywsgi import WSGIServer from flask import Flask, request, jsonify  app = Flask(__name__)  @app.route('/api', methods=['GET']) def api():     return jsonify({'message': 'Hello, Habr!'})  if __name__ == '__main__':     http_server = WSGIServer(('0.0.0.0', 5000), app)     print("Server running on http://0.0.0.0:5000")     http_server.serve_forever()

monkey.patch_all() используется для кооперативного патчинга стандартных библиотек, чтобы сделать их совместимыми с Gevent.

WSGIServer из gevent.pywsgi создает веб-сервер, который может обслуживать несколько запросов одновременно.

Веб-скрапинг с использованием Gevent

Гео-скрапинг сайтов для извлечения данных параллельно:

import gevent from gevent import monkey monkey.patch_all()  import requests from bs4 import BeautifulSoup  def fetch(url):     response = requests.get(url)     soup = BeautifulSoup(response.text, 'html.parser')     title = soup.find('title').text     print(f"Title: {title} from {url}")  urls = [     'http://www.google.com',     'http://www.example.com',     'http://www.python.org' ]  jobs = [gevent.spawn(fetch, url) for url in urls] gevent.joinall(jobs)

monkey.patch_all() для кооперативного выполнение сетевых запросов.

requests используется для отправки HTTP-запросов.

BeautifulSoup используется для парсинга HTML и извлечения данных.

Асинхронные микросервисы с использованием Gevent

Создадим микросервис, который выполняет асинхронные операции ввода-вывода:

from gevent import monkey monkey.patch_all()  import gevent from gevent.queue import Queue from flask import Flask, jsonify  app = Flask(__name__) queue = Queue()  @app.route('/add_task', methods=['POST']) def add_task():     task = request.json.get('task')     queue.put(task)     return jsonify({'status': 'Task added'}), 201  @app.route('/get_task', methods=['GET']) def get_task():     try:         task = queue.get_nowait()     except gevent.queue.Empty:         task = None     return jsonify({'task': task})  def worker():     while True:         task = queue.get()         if task:             print(f"Processing task: {task}")             gevent.sleep(2)  if __name__ == '__main__':     workers = [gevent.spawn(worker) for _ in range(3)]     http_server = WSGIServer(('0.0.0.0', 5000), app)     print("Microservice running on http://0.0.0.0:5000")     http_server.start()     gevent.joinall(workers + [http_server])

Создается очередь задач Queue, в которую добавляются задачи через HTTP-запросы.

Рабочие worker обрабатывают задачи из очереди параллельно.

Управление потоками данных в реальном времени

Создадим сервис для обработки данных в реальном времени, например как чат или игровая комната:

from gevent import monkey monkey.patch_all()  import gevent from gevent.queue import Queue from flask import Flask, request, jsonify  app = Flask(__name__) message_queues = {}  @app.route('/send_message', methods=['POST']) def send_message():     username = request.json.get('username')     message = request.json.get('message')     if username in message_queues:         message_queues[username].put_nowait(message)         return jsonify({'status': 'Message sent'}), 201     else:         return jsonify({'error': 'User not connected'}), 404  @app.route('/receive_messages/<username>', methods=['GET']) def receive_messages(username):     if username not in message_queues:         message_queues[username] = Queue()     messages = []     try:         while True:             message = message_queues[username].get_nowait()             messages.append(message)     except gevent.queue.Empty:         pass     return jsonify({'messages': messages})  if __name__ == '__main__':     http_server = WSGIServer(('0.0.0.0', 5000), app)     print("Real-time messaging service running on http://0.0.0.0:5000")     http_server.serve_forever()

Создали очереди сообщений для каждого юзера.

Сообщения добавляются в соответствующую очередь и могут быть извлечены через HTTP-запросы.

---

Подробнее с библиотекой можно ознакомиться здесь.

В завершение статьи хочу пригласить вас на бесплатный вебинар курса Highload Architect. Регистрация доступна по ссылке.