Многопоточность и параллелизм в Python: потоки, процессы и гринлеты

import threading
import time
 
def count():
    for i in range(1_000_000):
        pass
 
# Секвенциальный запуск
start = time.time()
count(); count()
print(f"Последовательное выполнение: {time.time() - start}")
 
# "Конкурентное" исполнение потоками
start = time.time()
t1 = threading.Thread(target=count)
t2 = threading.Thread(target=count)
t1.start(); t2.start(); t1.join(); t2.join()
print(f"Многопоточное выполнение: {time.time() - start}")

import multiprocessing as mp
 
# Реально параллельное выполнение
start = time.time()
p1 = mp.Process(target=count)
p2 = mp.Process(target=count)
p1.start(); p2.start(); p1.join(); p2.join()
print(f"Многопроцессное выполнение: {time.time() - start}")

import threading
import time
 
shared_resource = 0
 
def increment_slowly():
global shared_resource
local_copy = shared_resource
# Имитация сложной обработки
time.sleep(0.1)
shared_resource = local_copy + 1
 
threads = []
for i in range(10):
t = threading.Thread(target=increment_slowly)
threads.append(t)
t.start()
 
for t in threads:
t.join()
 
print(f"Результат: {shared_resource}")  # Ожидаем 10, но получим...?

# Правильный вариант с использованием блокировки
lock = threading.Lock()
 
def increment_safely():
global shared_resource
with lock:  # Аккуратно блокируем ресурс
    local_copy = shared_resource
    time.sleep(0.1)
    shared_resource = local_copy + 1

import multiprocessing as mp
 
def intensive_calculation(number):
return sum(i * i for i in range(number))
 
if __name__ == '__main__':  # Это критически важно для Windows!
pool = mp.Pool(processes=4)  # Создаём пул из 4 процессов
results = pool.map(intensive_calculation, [10**6, 10**6, 10**6, 10**6])
pool.close()
pool.join()
print(results)

import gevent
from gevent import socket
 
def fetch(url):
print(f'Начинаем загрузку {url}')
sock = socket.socket()
sock.connect((url, 80))
sock.send(b'GET / HTTP/1.0\r\n\r\n')
sock.recv(1024)  # Тут гринлет "уступает" время другим
print(f'Загрузка {url} завершена')
 
# Создаем и запускаем гринлеты
urls = ['www.google.com', 'www.example.com', 'www.python.org']
jobs = [gevent.spawn(fetch, url) for url in urls]
gevent.joinall(jobs)

from multiprocessing import Process, Queue
import threading
import gevent
 
def process_worker(queue):
    # Запускаем несколько потоков внутри процесса
    threads = [threading.Thread(target=thread_worker, args=(i,)) for i in range(3)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    # Отправляем результат обратно
    queue.put("Процесс завершил работу")
 
def thread_worker(id):
    # Внутри потока запускаем несколько гринлетов
    jobs = [gevent.spawn(greenlet_worker, f"{id}-{j}") for j in range(2)]
    gevent.joinall(jobs)
 
def greenlet_worker(id):
    print(f"Гринлет {id} выполняет задачу")
 
if __name__ == '__main__':
    q = Queue()
    proc = Process(target=process_worker, args=(q,))
    proc.start()
    print(q.get())  # Ждём результата от процесса
    proc.join()

import time
import threading
import multiprocessing
import gevent
 
def io_bound(n):
    # Имитируем операцию чтения из сети
    time.sleep(0.1)
    return n
 
def cpu_bound(n):
    # Имитируем сложные вычисления
    return sum(i*i for i in range(n))
 
def benchmark(func, tasks, worker_type):
    start = time.time()
    if worker_type == "sequential":
        results = [func(i) for i in tasks]
    # ... другие случаи реализации для разных типов конкурентности
    end = time.time()
    return end - start

# Упрощенная имитация работы GIL
import threading
import time
 
gil_lock = threading.Lock()
 
def thread_function(id):
    with gil_lock:  # Захватываем GIL
        print(f"Поток {id} захватил GIL")
        # Выполняем байткод Python
        sum(i for i in range(10**7))
        print(f"Поток {id} освобождает GIL")

import threading
import time
import sys
 
# Можем настроить, как часто потоки будут пытаться передать GIL
sys.setswitchinterval(0.005)  # Каждые 5 миллисекунд
 
def cpu_bound_task(n):
    start = time.time()
    # Этот цикл захватит GIL
    result = sum(i*i for i in range(n))
    print(f"Задача заняла {time.time() - start:.4f} секунд")
    return result
 
# Запускаем две задачи в разных потоках
t1 = threading.Thread(target=cpu_bound_task, args=(10**7,))
t2 = threading.Thread(target=cpu_bound_task, args=(10**7,))
 
start_total = time.time()
t1.start(); t2.start()
t1.join(); t2.join()
print(f"Общее время выполнения: {time.time() - start_total:.4f} секунд")

import threading
import time
 
def count_to_large_number(n):
    for _ in range(n):
        pass
 
# Однопоточное выполнение
start = time.time()
count_to_large_number(10[B]8)
count_to_large_number(10[/B]8)
single_thread_time = time.time() - start
print(f"Однопоточное время: {single_thread_time:.2f} секунд")
 
# Многопоточное выполнение
start = time.time()
t1 = threading.Thread(target=count_to_large_number, args=(10**8,))
t2 = threading.Thread(target=count_to_large_number, args=(10**8,))
t1.start(); t2.start()
t1.join(); t2.join()
multi_thread_time = time.time() - start
print(f"Многопоточное время: {multi_thread_time:.2f} секунд")
 
print(f"Многопоточный код работает в {multi_thread_time/single_thread_time:.2f} раза {'быстрее' if multi_thread_time < single_thread_time else 'медленнее'}")

import threading
 
counter = 0
counter_lock = threading.Lock()
 
def increment_counter(amount):
global counter
with counter_lock:  # Атомарный захват и последующее освобождение
    temp = counter
    temp += amount
    counter = temp
 
threads = []
for i in range(100):
    t = threading.Thread(target=increment_counter, args=(1,))
    threads.append(t)
    t.start()
 
for t in threads:
    t.join()
 
print(f"Итоговый счётчик: {counter}")  # Всегда будет 100!

import threading
import time
 
# Создаём семафор, который пропустит максимум 3 потока одновременно
pool_sema = threading.Semaphore(value=3)
 
def worker(id):
print(f"Поток {id} ждёт доступа к пулу ресурсов")
with pool_sema:
    # Крутая фича: с семафором можно также использовать конструкцию with
    print(f"Поток {id} получил ресурс")
    time.sleep(1)  # Имитируем долгую работу
    print(f"Поток {id} освободил ресурс")
 
threads = [threading.Thread(target=worker, args=(i,)) for i in range(10)]
for t in threads:
    t.start()

import threading
import time
import random
 
# Событие, которое будет установлено, когда все гонщики будут готовы
all_ready = threading.Event()
# Событие, сигнализирующее старт гонки
start_race = threading.Event()
results = []
results_lock = threading.Lock()
 
def racer(name):
# Имитируем подготовку к гонке
preparation_time = random.uniform(0.1, 1.0)
time.sleep(preparation_time)
print(f"{name} готов к гонке!")
    
all_ready.set()  # Сообщаем, что этот гонщик готов
start_race.wait()  # Ждём сигнала старта
    
# Гонка началась
race_time = random.uniform(1.0, 3.0)
time.sleep(race_time)
    
with results_lock:
    results.append((name, race_time))
    
print(f"{name} финишировал за {race_time:.2f} секунд!")
 
# Создаём потоки-гонщики
racers = ["Алиса", "Боб", "Чарли", "Дейв"]
threads = [threading.Thread(target=racer, args=(r,)) for r in racers]
 
# Запускаем всех гонщиков
for t in threads:
    t.start()
 
# Ждём, пока все гонщики будут готовы
all_ready.wait()
print("Все гонщики готовы! Начинаем гонку через...")
for i in range(3, 0, -1):
    print(i)
    time.sleep(1)
print("СТАРТ!")
 
# Даём сигнал к началу гонки
start_race.set()
 
# Ждём завершения всех потоков
for t in threads:
    t.join()
 
print("Результаты гонки:")
for name, time in sorted(results, key=lambda x: x[1]):
    print(f"{name}: {time:.2f} секунд")

from multiprocessing import Process, Queue
import time
 
def producer(q):
    for i in range(5):
        item = f"элемент-{i}"
        q.put(item)
        print(f"Производитель добавил {item}")
        time.sleep(0.5)
    # Сигнализируем о завершении
    q.put(None)
 
def consumer(q):
    while True:
        item = q.get()
        if item is None:  # Сигнал о завершении
            break
        print(f"Потребитель получил {item}")
        time.sleep(1.0)
 
if __name__ == "__main__":
    q = Queue()
    p1 = Process(target=producer, args=(q,))
    p2 = Process(target=consumer, args=(q,))
    
    p1.start()
    p2.start()
    
    p1.join()
    p2.join()

from multiprocessing import Process, Pipe, Value, Array
import time
 
def sender(conn, value, array):
    # Отправляем через пайп
    for i in range(5):
        conn.send(f"сообщение-{i}")
        time.sleep(0.5)
    conn.close()
    
    # Модифицируем разделяемые объекты
    with value.get_lock():
        value.value += 100
        
    for i in range(len(array)):
        array[i] *= 2
 
def receiver(conn):
    while True:
        try:
            msg = conn.recv()
            print(f"Получено: {msg}")
        except EOFError:
            print("Канал связи закрыт")
            break
 
if __name__ == "__main__":
    # Создаём пайп для общения
    parent_conn, child_conn = Pipe()
    
    # Создаём разделяемые объекты
    shared_value = Value('i', 50)  # 'i' - тип int
    shared_array = Array('d', [1.0, 2.0, 3.0])  # 'd' - тип double
    
    p1 = Process(target=sender, args=(child_conn, shared_value, shared_array))
    p2 = Process(target=receiver, args=(parent_conn,))
    
    p1.start()
    p2.start()
    
    p1.join()
    p2.join()
    
    print(f"Итоговое значение: {shared_value.value}")
    print(f"Итоговый массив: {list(shared_array)}")

from greenlet import greenlet
 
def ping():
    print("пинг")
    pong_gl.switch()
    print("пинг")
    pong_gl.switch()
 
def pong():
    print("понг")
    ping_gl.switch()
    print("понг")
 
# Создаём гринлеты
ping_gl = greenlet(ping)
pong_gl = greenlet(pong)
 
# Стартуем пинг-понг
ping_gl.switch()

import eventlet
from eventlet import spawn
 
def fetch(url):
    print(f"Начинаем загрузку {url}")
    # Автоматически уступит управление во время ожидания
    data = eventlet.urllib.request.urlopen(url).read()
    print(f"Завершена загрузка {url} ({len(data)} байт)")
    return len(data)
 
# Патчим стандартные библиотеки
eventlet.monkey_patch()
 
# Создаём пул гринлетов
pool = eventlet.GreenPool(size=4)
urls = ['http://example.com', 'http://python.org', 'http://google.com']
 
# Запускаем задачи
results = pool.imap(fetch, urls)
for url, size in zip(urls, results):
    print(f"{url} вернул {size} байт")

import gevent
from gevent import monkey; monkey.patch_all()
import requests
 
def download(url):
    print(f"Скачиваем {url}")
    response = requests.get(url)
    return f"{url}: {len(response.content)} байт"
 
# Создаём и запускаем гринлеты для каждого URL
urls = ["https://example.org", "https://python.org", "https://github.com"]
jobs = [gevent.spawn(download, url) for url in urls]
 
# Ждём завершения всех задач
gevent.joinall(jobs)
 
# Получаем результаты
for job in jobs:
    print(job.value)

# Упрощенный бенчмарк
import time
import gevent
import eventlet
 
def bench_gevent(n):
    start = time.time()
    # ... геврент имплементация ...
    end = time.time()
    return end - start
 
def bench_eventlet(n):
    start = time.time()
    # ... эвентлет имплементация ...
    end = time.time()
    return end - start
 
# Результаты на 1000 параллельных HTTP запросов
# gevent: ~1.2s
# eventlet: ~1.35s

import threading
import requests
from queue import Queue
 
def worker(url_queue, results):
while not url_queue.empty():
    url = url_queue.get()
    try:
        response = requests.get(url, timeout=5)
        results[url] = len(response.content)
    except Exception as e:
        results[url] = f"Ошибка: {str(e)}"
    finally:
        url_queue.task_done()
 
# Подготовка данных
urls = ["https://python.org", "https://github.com", "https://stackoverflow.com"] * 10
url_queue = Queue()
for url in urls:
url_queue.put(url)
 
# Запуск потоков
results = {}
threads = []
for _ in range(10):  # 10 параллельных воркеров
t = threading.Thread(target=worker, args=(url_queue, results))
t.daemon = True  # Позволит программе завершиться, даже если поток еще работает
threads.append(t)
t.start()
 
# Ожидание завершения всех задач
url_queue.join()
 
# Вывод результатов
for url, size in results.items():
print(f"{url}: {size}")

import multiprocessing as mp
from PIL import Image, ImageFilter
import time
import os
 
def process_image(img_path):
img = Image.open(img_path)
result = img.filter(ImageFilter.GaussianBlur(radius=10))
save_path = f"processed_{os.path.basename(img_path)}"
result.save(save_path)
return save_path
 
if __name__ == "__main__":
img_paths = [f"image_{i}.jpg" for i in range(1, 20)]
 
 
# Последовательная обработка
start = time.time()
for path in img_paths:
process_image(path)
seq_time = time.time() - start
print(f"Последовательная обработка: {seq_time:.2f} секунд")
 
# Параллельная обработка
start = time.time()
with mp.Pool(processes=mp.cpu_count()) as pool:
results = pool.map(process_image, img_paths)
par_time = time.time() - start
print(f"Параллельная обработка: {par_time:.2f} секунд")
print(f"Ускорение: {seq_time/par_time:.2f}x")

import gevent
from gevent import monkey; monkey.patch_all()
import requests
import time
 
def fetch_site(url):
print(f"Начало запроса к {url}")
start = time.time()
response = requests.get(url)
latency = time.time() - start
print(f"Завершён запрос к {url}: {len(response.content)} байт, задержка {latency:.2f}c")
return (url, len(response.content), latency)
 
# Список URL для тестирования
urls = ["https://www.google.com", "https://www.github.com", "https://www.python.org",
       "https://www.reddit.com", "https://www.wikipedia.org"] * 5
 
# Запуск через гринлеты
start = time.time()
jobs = [gevent.spawn(fetch_site, url) for url in urls]
gevent.joinall(jobs)
results = [job.value for job in jobs]
end = time.time()
 
# Анализ результатов
total_bytes = sum(r[1] for r in results)
avg_latency = sum(r[2] for r in results) / len(results)
print(f"\nВсего загружено: {total_bytes/1024:.1f} Кб")
print(f"Средняя задержка: {avg_latency:.2f} секунд")
print(f"Общее время выполнения: {end - start:.2f} секунд")
print(f"Эффективная скорасть: {len(urls)/(end - start):.1f} запросов/сек")

import threading
import multiprocessing as mp
import gevent
import time
import requests
import math
from functools import partial
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
 
def benchmark(name, func, args, repeats=3):
times = []
for i in range(repeats):
    start = time.time()
    result = func(*args)
    elapsed = time.time() - start
    times.append(elapsed)
    print(f"{name} ({i+1}/{repeats}): {elapsed:.3f}s")
print(f"{name} (avg): {sum(times)/len(times):.3f}s")
return sum(times)/len(times)
 
# Задача, интенсивная по I/O: загрузка веб-страниц
def fetch_all_sync(urls):
return [requests.get(url).text for url in urls]
 
def fetch_all_threading(urls):
def fetch(url):
    return requests.get(url).text
with ThreadPoolExecutor(max_workers=10) as executor:
    return list(executor.map(fetch, urls))
 
def fetch_all_multiprocessing(urls):
def fetch(url):
    return requests.get(url).text
with ProcessPoolExecutor(max_workers=10) as executor:
    return list(executor.map(fetch, urls))
 
def fetch_all_gevent(urls):
from gevent import monkey; monkey.patch_all()
jobs = [gevent.spawn(requests.get, url) for url in urls]
gevent.joinall(jobs)
return [job.value.text for job in jobs]
 
# Задача, интенсивная по CPU: вычисление простых чисел
def is_prime(n):
if n < 2:
    return False
for i in range(2, int(math.sqrt(n))+1):
    if n % i == 0:
        return False
return True
 
def count_primes_sync(range_tuple):
start, end = range_tuple
return sum(1 for i in range(start, end) if is_prime(i))
 
def count_primes_threading(range_tuple, num_threads=10):
start, end = range_tuple
chunk_size = (end - start) // num_threads
chunks = [(start + i * chunk_size, start + (i+1) * chunk_size) 
         for i in range(num_threads)]
 
 
with ThreadPoolExecutor(max_workers=num_threads) as executor:
    results = list(executor.map(count_primes_sync, chunks))
return sum(results)
 
def count_primes_multiprocessing(range_tuple, num_processes=None):
start, end = range_tuple
if num_processes is None:
    num_processes = mp.cpu_count()
    
chunk_size = (end - start) // num_processes
chunks = [(start + i * chunk_size, start + (i+1) * chunk_size) 
         for i in range(num_processes)]
    
with ProcessPoolExecutor(max_workers=num_processes) as executor:
    results = list(executor.map(count_primes_sync, chunks))
return sum(results)
 
# Тестирование на I/O-bound задаче
urls = ["http://example.com", "http://python.org", "http://github.com"] * 5
print("=== I/O-Bound Benchmark ===")
time_io_sync = benchmark("Sync I/O", fetch_all_sync, (urls,))
time_io_thread = benchmark("Threaded I/O", fetch_all_threading, (urls,))
time_io_mp = benchmark("Multiprocess I/O", fetch_all_multiprocessing, (urls,))
time_io_gevent = benchmark("Gevent I/O", fetch_all_gevent, (urls,))
 
# Тестирование на CPU-bound задаче
prime_range = (1, 500000)
print("\n=== CPU-Bound Benchmark ===")
time_cpu_sync = benchmark("Sync CPU", count_primes_sync, (prime_range,))
time_cpu_thread = benchmark("Threaded CPU", count_primes_threading, (prime_range,))
time_cpu_mp = benchmark("Multiprocess CPU", count_primes_multiprocessing, (prime_range,))
 
# Вывод результатов сравнения
print("\n=== Сравнение производительности ===")
print(f"I/O-bound задача:")
print(f"Threading vs Sync: {time_io_sync/time_io_thread:.2f}x")
print(f"Multiprocessing vs Sync: {time_io_sync/time_io_mp:.2f}x")
print(f"Gevent vs Sync: {time_io_sync/time_io_gevent:.2f}x")
 
print(f"\nCPU-bound задача:")
print(f"Threading vs Sync: {time_cpu_sync/time_cpu_thread:.2f}x")
print(f"Multiprocessing vs Sync: {time_cpu_sync/time_cpu_mp:.2f}x")

# ПЛОХО: Ожидание ускорения CPU-задачи через потоки
import threading
 
def cpu_heavy_task():
result = sum(i*i for i in range(10**7))
return result
 
# Создаём 4 потока для "параллельной" работы
threads = [threading.Thread(target=cpu_heavy_task) for _ in range(4)]
for t in threads:
t.start()
for t in threads:
t.join()
 
# В реальности ваш код будет выполняться последовательно из-за GIL,
# но с дополнительными накладными расходами на управление потоками

# ПЛОХО: Необезопасенная работа с общими данными
import threading
counter = 0
 
def increment():
global counter
for _ in range(100000):
    # Этот, казалось бы, атомарный стейтмент в действительности 
    # состоит из нескольких операций (чтение, инкремент, запись)
    counter += 1  # <-- Опасная гонка данных!
 
threads = [threading.Thread(target=increment) for _ in range(5)]
for t in threads:
t.start()
for t in threads:
t.join()
 
print(f"Ожидаемый результат: 500000, Фактический: {counter}")
# Почти наверняка вы получите число меньше 500000

# ПЛОХО: Потенциальная взаимная блокировка
import threading
 
lock_a = threading.Lock()
lock_b = threading.Lock()
 
def task_one():
with lock_a:
# Если task_two уже захватил lock_b, мы останемся здесь навечно
print("Task one got lock A, waiting for lock B...")
with lock_b:
    print("Task one working with both resources")
 
def task_two():
with lock_b:
# Если task_one уже захватил lock_a, мы останемся здесь навечно
print("Task two got lock B, waiting for lock A...")
with lock_a:
    print("Task two working with both resources")
 
# Запуск потоков, которые могут привести к взаимоблокировке
t1 = threading.Thread(target=task_one)
t2 = threading.Thread(target=task_two)
t1.start(); t2.start()
t1.join(); t2.join()

# ПЛОХО: Чрезмерная многопоточность для простой задачи
import threading
 
def simple_task(n):
return n * 2
 
# Избыточная многопоточность для тривиальной операции
data = list(range(10))
results = [None] * len(data)
 
def process_item(idx, value):
results[idx] = simple_task(value)
 
threads = [threading.Thread(target=process_item, args=(i, n)) for i, n in enumerate(data)]
for t in threads:
t.start()
for t in threads:
t.join()
 
# Накладные расходы на создание потоков превышают выигрыш от параллельности
# для таких легковесных операций

# ПЛОХО: Потоки-демоны без контроля
import threading
import time
 
def background_task():
while True:
    print("Working...")
    time.sleep(1)
 
# Создаем и запускаем поток
t = threading.Thread(target=background_task)
t.start()
# ... основной код ...
[H2]Завершаем программу, но поток остается висеть в памяти[/H2]
 
# ХОРОШО: Используем daemon=True или явно управляем жизненным циклом
t = threading.Thread(target=background_task, daemon=True)
t.start()
# При завершении программы daemon-потоки будут автоматически уничтожены

# ХОРОШО: Правильный выбор инструмента в зависимости от задачи
import concurrent.futures
import time
 
def io_bound_task(url):
# Операция, ограниченная вводом-выводом
import requests
return requests.get(url).text[:100]  # Возвращаем первые 100 символов
 
def cpu_bound_task(n):
# Вычислительно интенсивная операция
return sum(i*i for i in range(n))
 
# Для I/O-bound задач используем ThreadPoolExecutor
urls = ["https://example.com", "https://python.org", "https://github.com"] * 3
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
results = list(executor.map(io_bound_task, urls))
 
# Для CPU-bound задач используем ProcessPoolExecutor
numbers = [10**6, 10**6, 10**6, 10**6]
with concurrent.futures.ProcessPoolExecutor() as executor:
results = list(executor.map(cpu_bound_task, numbers))

import cProfile
import pstats
from io import StringIO
 
def my_function():
# Ваш код здесь
pass
 
# Профилируем функцию
pr = cProfile.Profile()
pr.enable()
my_function()
pr.disable()
 
# Анализируем результаты профилирования
s = StringIO()
ps = pstats.Stats(pr, stream=s).sort_stats('cumulative')
ps.print_stats()
print(s.getvalue())

import threading
import requests
import time
 
def download_site(url):
print(f"Начало загрузки {url}")
response = requests.get(url)
print(f"Завершена загрузка {url}, получено {len(response.content)} байт")
 
sites = [
"https://www.python.org",
"https://www.google.com",
"https://www.github.com",
] * 3  # повторим каждый URL по 3 раза
 
# Последовательный подход
start_time = time.time()
for site in sites:
download_site(site)
print(f"Последовательное время: {time.time() - start_time:.2f} секунд")
 
# Многопоточный подход
start_time = time.time()
threads = []
for site in sites:
thread = threading.Thread(target=download_site, args=(site,))
threads.append(thread)
thread.start()
 
for thread in threads:
thread.join()
print(f"Многопоточное время: {time.time() - start_time:.2f} секунд")

import asyncio
import aiohttp
import time
 
async def download_site(session, url):
print(f"Начало загрузки {url}")
async with session.get(url) as response:
    content = await response.read()
    print(f"Завершена загрузка {url}, получено {len(content)} байт")
 
async def download_all_sites(sites):
async with aiohttp.ClientSession() as session:
    tasks = [download_site(session, site) for site in sites]
    await asyncio.gather(*tasks)
 
sites = [
"https://www.python.org",
"https://www.google.com",
"https://www.github.com",
] * 3
 
# Измеряем время выполнения
start_time = time.time()
asyncio.run(download_all_sites(sites))
print(f"Асинхронное время: {time.time() - start_time:.2f} секунд")

import gevent
from gevent import monkey
import requests
import time
 
# Патчим стандартные библиотеки для автоматического переключения гринлетов
monkey.patch_all()
 
def download_site(url):
print(f"Начало загрузки {url}")
response = requests.get(url)
print(f"Завершена загрузка {url}, получено {len(response.content)} байт")
return len(response.content)
 
sites = [
"https://www.python.org",
"https://www.google.com",
"https://www.github.com",
] * 3
 
# Запускаем через гринлеты
start_time = time.time()
jobs = [gevent.spawn(download_site, site) for site in sites]
gevent.joinall(jobs)
print(f"Gevent-время: {time.time() - start_time:.2f} секунд")

import multiprocessing
import time
 
def cpu_bound_task(number):
"""Вычислительно-интенсивная задача"""
total = 0
for i in range(number):
    total += i [B] 2
return total
 
numbers = [10[/B]7, 10**7, 10**7, 10**7]  # Список входных данных
 
# Последовательный подход
start_time = time.time()
for number in numbers:
cpu_bound_task(number)
print(f"Последовательное время: {time.time() - start_time:.2f} секунд")
 
# Многопроцессный подход
if __name__ == "__main__":  # Это очень важно для Windows!
    start_time = time.time()
    with multiprocessing.Pool() as pool:
        pool.map(cpu_bound_task, numbers)
    print(f"Многопроцессное время: {time.time() - start_time:.2f} секунд")

import numpy as np
import threading
import time
 
def numpy_calculation(size):
# NumPy освобождает GIL во время выполнения этих операций
matrix = np.random.rand(size, size)
result = np.dot(matrix, matrix.T)
return result
 
# Последовательное выполнение
sizes = [1000, 1000, 1000, 1000]
start_time = time.time()
for size in sizes:
numpy_calculation(size)
print(f"Последовательное время: {time.time() - start_time:.2f} секунд")
 
# Многопоточное выполнение
start_time = time.time()
threads = []
for size in sizes:
thread = threading.Thread(target=numpy_calculation, args=(size,))
threads.append(thread)
thread.start()
 
for thread in threads:
thread.join()
print(f"Многопоточное время: {time.time() - start_time:.2f} секунд")

import multiprocessing as mp
from multiprocessing import Queue
import time
import random
 
def worker(task_queue, result_queue):
"""Процесс-работник, который выполняет задачи из очереди"""
while True:
    try:
        task_id, task_type, data = task_queue.get(timeout=1)
        if task_type == "io":
            # Имитация I/O задачи
            time.sleep(random.random())
            result = f"I/O выполнено для {data}"
        elif task_type == "cpu":
            # Имитация CPU задачи
            result = sum(i * i for i in range(data))
        else:
            result = "Неизвестный тип задачи"
        
        result_queue.put((task_id, result))
    except:
        # Если очередь пуста 1 секунду, завершаем работу
        break
 
def main():
# Создаём очереди задач и результатов
task_queue = Queue()
result_queue = Queue()
 
# Загружаем некоторые задачи в очередь
tasks = [
    (1, "io", "https://example.com"),
    (2, "cpu", 10**6),
    (3, "io", "file.txt"),
    (4, "cpu", 10**7),
    (5, "io", "https://python.org"),
    (6, "cpu", 10**5),
]
for task in tasks:
    task_queue.put(task)
 
# Запускаем рабочие процессы
num_workers = mp.cpu_count()
workers = [mp.Process(target=worker, args=(task_queue, result_queue)) 
          for _ in range(num_workers)]
for w in workers:
    w.start()
 
# Собираем результаты
results = {}
while len(results) < len(tasks):
    task_id, result = result_queue.get()
    results[task_id] = result
    print(f"Задача {task_id} завершена: {result}")
 
# Ждём, пока все процессы завершатся
for w in workers:
    w.join()
 
if __name__ == "__main__":
main()

import multiprocessing as mp
import threading
import time
import random
 
def io_task(url):
# Имитация I/O-задачи
time.sleep(random.random() * 0.1)
return f"Загружено {url}"
 
def cpu_and_io_task(batch_id, urls):
"""Эта функция выполняет CPU-работу и создаёт потоки для I/O-задач"""
# CPU-интенсивная часть
result = sum(i * i for i in range(10**6))
 
 
# I/O-часть, распараллеленная потоками
io_results = []
threads = []
for url in urls:
    t = threading.Thread(target=lambda u: io_results.append(io_task(u)), args=(url,))
    threads.append(t)
    t.start()
 
for t in threads:
    t.join()
    
return (batch_id, result, io_results)
 
if __name__ == "__main__":
# Подготовка данных для задач
batches = [
    (1, ["url1", "url2", "url3"]),
    (2, ["url4", "url5", "url6"]),
    (3, ["url7", "url8", "url9", "url10"]),
    (4, ["url11", "url12"])
]
 
# Создаём пул процессов
with mp.Pool(processes=mp.cpu_count()) as pool:
    # map_async не блокирует основной поток
    result = pool.starmap(cpu_and_io_task, [(bid, urls) for bid, urls in batches])
    
    for batch_id, cpu_result, io_results in result:
        print(f"Batch {batch_id}:")
        print(f"  CPU result: {cpu_result}")
        print(f"  I/O results: {io_results}")