Как определить количество CPU в Python: 3 эффективных метода

Для кого эта статья:

• Python-разработчики, работающие с высоконагруженными приложениями
• Специалисты, занимающиеся системным программированием и оптимизацией производительности

• Исследователи и студенты, изучающие распределенные системы и параллельные вычисления

  При разработке высоконагруженных приложений или систем обработки данных знание точного количества доступных CPU критически важно для максимальной производительности. Неоптимальное распределение задач может привести либо к простою ресурсов, либо к перегрузке системы. Python, благодаря своей гибкости и богатой экосистеме библиотек, предоставляет несколько изящных способов определения количества процессоров — от встроенных модулей до специализированных решений. Разберем три наиболее эффективных метода, которые помогут вашему коду адаптироваться к любой аппаратной конфигурации. 🚀

Хотите писать эффективный код, который адаптируется к аппаратным ресурсам? Курс Обучение Python-разработке от Skypro погружает вас в тонкости системного программирования и оптимизации. Вы научитесь создавать масштабируемые решения, учитывающие количество доступных CPU, эффективно распараллеливать задачи и добиваться максимальной производительности вашего кода на любых конфигурациях оборудования.

Почему важно определять количество CPU для оптимизации

Точное знание количества доступных процессорных ядер позволяет значительно повысить эффективность вычислительных задач. Программы, неспособные определить и адаптироваться к аппаратной конфигурации, рискуют либо недоиспользовать ресурсы, либо создать избыточную нагрузку, что приведет к замедлению всей системы. 💻

Алексей Смирнов, ведущий разработчик систем обработки данных

Однажды мы столкнулись с проблемой производительности нашего сервиса обработки аналитических данных. Несмотря на мощное оборудование, задачи выполнялись неприемлемо медленно. После анализа выяснилось, что система создавала фиксированное количество рабочих процессов, независимо от доступных ресурсов. На 4-ядерных машинах производительность была приемлемой, но на 32-ядерных серверах большинство ресурсов простаивало.

После внедрения автоматического определения количества CPU и динамического распределения нагрузки производительность выросла почти в 7 раз на мощных серверах. Это был важный урок: программы должны "знать", с какими ресурсами они работают, особенно в гетерогенной инфраструктуре.

Вот ключевые причины, почему определение количества CPU критически важно:

• Параллельное выполнение задач может ускорить обработку в N раз, где N — количество доступных процессоров
• Оптимальное количество рабочих процессов обычно соотносится с количеством физических или логических ядер
• Избыточное создание потоков/процессов может привести к деградации производительности
• Кросс-платформенные приложения должны адаптироваться к различным конфигурациям оборудования
• Облачные приложения работают в средах с динамическим количеством доступных ресурсов

Теперь рассмотрим конкретные методы определения количества CPU, начиная с самого простого и универсального.

Метод №1: Определение CPU через multiprocessing в Python

Модуль multiprocessing из стандартной библиотеки Python — наиболее простой и элегантный способ определить количество доступных процессоров. Этот метод работает единообразно во всех основных операционных системах (Windows, Linux, macOS), что делает его идеальным выбором для кросс-платформенных приложений. 🔍

Базовый пример использования:

import multiprocessing

# Получение количества доступных CPU
cpu_count = multiprocessing.cpu_count()
print(f"Доступно CPU: {cpu_count}")

# Создание пула процессов, оптимизированного под имеющиеся ресурсы
def process_data(data):
return data * data

if __name__ == '__main__':
data = list(range(10000000))
with multiprocessing.Pool(processes=cpu_count) as pool:
result = pool.map(process_data, data)

Метод cpu_count() возвращает количество логических процессоров, доступных системе. Для современных процессоров с технологией Hyper-Threading это значение может быть вдвое больше количества физических ядер.

Преимущества этого метода:

• Входит в стандартную библиотеку — не требует установки дополнительных пакетов
• Высокая кросс-платформенная совместимость
• Простота использования — всего одна строка кода
• Интеграция с другими функциями модуля multiprocessing
• Возвращает именно то количество процессоров, которое Python может использовать

Однако у этого метода есть несколько ограничений:

• Возвращает None, если не может определить количество процессоров
• Не различает физические ядра и логические процессоры
• Не предоставляет детальной информации о CPU (частота, модель и т.д.)

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

import multiprocessing
import time
import math

def calculate_pi_portion(start, end, step):
"""Расчет числа π методом Лейбница на заданном интервале"""
result = 0
for i in range(start, end):
result += 4 * ((-1) ** i) / (2 * i + 1)
return result

def calculate_pi_parallel(num_steps):
"""Параллельный расчет числа π"""
num_cpus = multiprocessing.cpu_count()
pool = multiprocessing.Pool(processes=num_cpus)

# Разделяем работу между CPU
step_per_cpu = num_steps // num_cpus
ranges = [(i * step_per_cpu, (i + 1) * step_per_cpu, num_steps) 
for i in range(num_cpus)]

# Последнему процессу отдаем остаток шагов
ranges[-1] = (ranges[-1][0], num_steps, num_steps)

results = pool.starmap(calculate_pi_portion, ranges)
pool.close()
pool.join()

# Суммируем результаты со всех процессоров
return sum(results)

if __name__ == "__main__":
NUM_STEPS = 100000000

start_time = time.time()
pi = calculate_pi_parallel(NUM_STEPS)
end_time = time.time()

print(f"π ≈ {pi}")
print(f"Расчет на {multiprocessing.cpu_count()} CPU занял {end_time – start_time:.2f} секунд")

Метод №2: Получение данных о процессорах с помощью os

Модуль os предоставляет низкоуровневый доступ к функциям операционной системы, включая информацию о доступных процессорах. Этот подход полезен, когда требуются платформозависимые оптимизации или более подробные сведения о конфигурации системы. 🖥️

Дмитрий Волков, DevOps-инженер

Мы поддерживали старую систему мониторинга инфраструктуры, написанную на Python. На Linux-серверах всё работало идеально, но когда мы начали добавлять Windows-машины, скрипты начали выдавать неточные данные о загрузке CPU.

Проблема была в коде, который получал количество процессоров через os.sysconf() — метод, специфичный для UNIX-систем. Для решения пришлось написать условную логику, определяющую ОС и выбирающую подходящий метод получения данных о CPU. Для Linux использовался os.sysconf('SCNPROCESSORSONLN'), а для Windows — впоследствии добавленный os.cpu_count() (в новых версиях Python).

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

В зависимости от операционной системы, модуль os предлагает разные методы для определения количества процессоров:

import os
import platform

def get_cpu_count_os():
"""Определение количества CPU с использованием модуля os"""

system = platform.system()

if system == "Linux" or system == "Darwin": # Linux или macOS
try:
# Количество доступных процессоров
cpu_count = os.sysconf("SC_NPROCESSORS_ONLN")
return cpu_count
except (ValueError, OSError, AttributeError):
pass

# Начиная с Python 3.4 есть os.cpu_count()
if hasattr(os, "cpu_count"):
return os.cpu_count()

# Если предыдущие методы не сработали
return None

# Проверка работы функции
print(f"Операционная система: {platform.system()}")
print(f"Количество CPU: {get_cpu_count_os()}")

Важно отметить, что os.cpu_count() (доступен с Python 3.4) выполняет ту же функцию, что и multiprocessing.cpu_count(), и фактически является предпочтительным методом в новом коде.

Однако модуль os предоставляет и другие возможности для получения системной информации:

Особый интерес представляет функция os.sched_getaffinity() в Linux, которая позволяет определить, какие процессоры доступны конкретному процессу. Это полезно в системах с управлением ресурсами, где процесс может быть ограничен подмножеством доступных CPU:

import os

def get_process_cpu_affinity():
"""Определение CPU, доступных текущему процессу (только для Linux)"""
try:
# Получаем маску процессорной принадлежности для текущего процесса
cpu_affinity = os.sched_getaffinity(0)
return {
"available_cpus": len(cpu_affinity),
"cpu_list": sorted(list(cpu_affinity))
}
except AttributeError:
return {"error": "Функция не поддерживается в данной ОС"}

# Проверка работы функции
affinity_info = get_process_cpu_affinity()
print(f"Информация о доступных процессору CPU: {affinity_info}")

Преимущества метода с использованием модуля os:

• Доступ к низкоуровневым системным функциям
• Различные методы для разных операционных систем
• Возможность получить более детальную информацию в некоторых ОС
• Входит в стандартную библиотеку Python

Недостатки:

• Требует дополнительной обработки для обеспечения кросс-платформенности
• Некоторые функции специфичны для определенных ОС
• Ограниченная информация о характеристиках процессоров

Метод №3: Расширенные возможности psutil для анализа CPU

Библиотека psutil (process and system utilities) предоставляет наиболее полный и гибкий инструментарий для получения информации о системных ресурсах, включая детальные сведения о процессорах. Этот метод идеален, когда требуется глубокий анализ или мониторинг CPU. 🔬

Для начала необходимо установить библиотеку:

pip install psutil

Базовый пример получения информации о CPU:

import psutil

# Количество логических CPU
logical_cpus = psutil.cpu_count()
print(f"Логических CPU: {logical_cpus}")

# Количество физических ядер (без учета Hyper-Threading/SMT)
physical_cpus = psutil.cpu_count(logical=False)
print(f"Физических ядер: {physical_cpus}")

# Загрузка CPU (в процентах)
cpu_usage = psutil.cpu_percent(interval=1)
print(f"Общая загрузка CPU: {cpu_usage}%")

# Загрузка каждого ядра (в процентах)
per_cpu_usage = psutil.cpu_percent(interval=1, percpu=True)
print(f"Загрузка по ядрам: {per_cpu_usage}")

Библиотека psutil позволяет получить гораздо больше информации о CPU, включая:

• Разделение на физические ядра и логические процессоры
• Текущую и историческую загрузку CPU
• Статистику времени, проведенного в различных состояниях (пользовательский режим, системный режим, простой)
• Детальную информацию о частотах процессора (на некоторых платформах)
• Температуру CPU (при наличии датчиков и на поддерживаемых платформах)

Рассмотрим более сложный пример, демонстрирующий расширенные возможности:

import psutil
import time
import json
from datetime import datetime

def get_detailed_cpu_info():
"""Получение детальной информации о процессорах"""

info = {
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"cpu_count": {
"logical": psutil.cpu_count(),
"physical": psutil.cpu_count(logical=False)
},
"cpu_usage": {
"total": psutil.cpu_percent(interval=1),
"per_cpu": psutil.cpu_percent(interval=1, percpu=True)
},
"cpu_times": {
"total": dict(psutil.cpu_times()._asdict()),
"per_cpu": [dict(t._asdict()) for t in psutil.cpu_times(percpu=True)]
},
"cpu_stats": dict(psutil.cpu_stats()._asdict())
}

# Получение частот CPU (поддерживается не на всех платформах)
try:
cpu_freq = psutil.cpu_freq()
if cpu_freq:
info["cpu_freq"] = {
"current": cpu_freq.current,
"min": cpu_freq.min,
"max": cpu_freq.max
}
except (AttributeError, NotImplementedError):
info["cpu_freq"] = {"error": "Не поддерживается"}

# Получение температуры CPU (поддерживается не на всех платформах)
try:
temps = psutil.sensors_temperatures()
if temps:
# Выбираем только датчики, связанные с CPU
cpu_temps = {}
for name, entries in temps.items():
if "cpu" in name.lower() or "core" in name.lower():
cpu_temps[name] = [{"label": e.label, "temp": e.current} 
for e in entries]
info["cpu_temp"] = cpu_temps
except (AttributeError, NotImplementedError):
info["cpu_temp"] = {"error": "Не поддерживается"}

return info

# Получение информации о CPU
cpu_info = get_detailed_cpu_info()
print(json.dumps(cpu_info, indent=2))

Для эффективного использования многопроцессорных систем psutil можно интегрировать с multiprocessing:

import psutil
import multiprocessing
import time
import numpy as np

def cpu_intensive_task(size):
"""Выполняет ресурсоемкую операцию с матрицами"""
matrix_a = np.random.rand(size, size)
matrix_b = np.random.rand(size, size)
result = np.dot(matrix_a, matrix_b)
return result.sum()

def monitor_cpu(stop_event, interval=0.5):
"""Мониторинг загрузки CPU в реальном времени"""
cpu_usage_history = []

while not stop_event.is_set():
usage = psutil.cpu_percent(percpu=True)
cpu_usage_history.append(usage)
time.sleep(interval)

return cpu_usage_history

if __name__ == "__main__":
# Определение оптимального количества рабочих процессов
physical_cores = psutil.cpu_count(logical=False)

# Используем 75% от доступных физических ядер
optimal_workers = max(1, int(physical_cores * 0.75))
print(f"Физических ядер: {physical_cores}")
print(f"Оптимальное количество рабочих процессов: {optimal_workers}")

# Создаем событие для остановки мониторинга
stop_monitoring = multiprocessing.Event()

# Запускаем мониторинг в отдельном процессе
monitoring_process = multiprocessing.Process(
target=monitor_cpu, 
args=(stop_monitoring,)
)
monitoring_process.start()

# Создаем пул процессов для вычислительных задач
with multiprocessing.Pool(processes=optimal_workers) as pool:
# Выполняем интенсивную задачу с разными размерами матриц
matrix_sizes = [500, 1000, 1500, 2000]
start_time = time.time()

results = pool.map(cpu_intensive_task, matrix_sizes)

end_time = time.time()

# Останавливаем мониторинг
stop_monitoring.set()
monitoring_process.join()

print(f"Задачи выполнены за {end_time – start_time:.2f} секунд")
print(f"Результаты: {results}")

Преимущества использования psutil:

• Наиболее полная информация о процессорах и их характеристиках
• Высокая кросс-платформенная совместимость
• Возможность мониторинга загрузки CPU в реальном времени
• Разделение на физические ядра и логические процессоры
• Доступ к дополнительным параметрам: частоте, температуре, статистике использования

Недостатки:

• Требует установки дополнительной библиотеки
• Некоторые функции могут быть недоступны на определенных платформах
• Избыточность для простых задач определения количества CPU

Сравнение методов и выбор оптимального решения

Выбор метода определения количества CPU зависит от конкретных требований проекта, целевых платформ и необходимой детализации информации. Сравним три рассмотренных подхода для различных сценариев использования. 📊

Рекомендации по выбору метода для различных сценариев:

• multiprocessing.cpu_count(): Идеален для большинства случаев, когда необходимо просто определить количество CPU для оптимизации параллельных вычислений. Это наиболее простой и универсальный метод, который следует использовать по умолчанию.

• os модуль: Подходит для низкоуровневых системных приложений или когда требуется платформозависимая информация. Полезен в старых версиях Python или при необходимости доступа к конкретным системным параметрам.

• psutil: Оптимален для мониторинга систем, анализа производительности или когда требуется детальная информация о процессорах. Незаменим при разработке инструментов системного анализа, оптимизации или для сложных сценариев балансировки нагрузки.

Пример гибкого определения количества процессоров с учетом контекста:

def get_optimal_workers(task_type="io_bound", safety_factor=0.75):
"""
Определение оптимального количества рабочих процессов
в зависимости от типа задачи и доступных ресурсов

Args:
task_type: Тип задачи ("io_bound" или "cpu_bound")
safety_factor: Коэффициент безопасности (0.0-1.0)

Returns:
int: Оптимальное количество рабочих процессов
"""
try:
import psutil
# Для CPU-интенсивных задач оптимальнее использовать физические ядра
if task_type == "cpu_bound":
cores = psutil.cpu_count(logical=False)
else:
# Для IO-интенсивных задач можно использовать логические ядра
cores = psutil.cpu_count(logical=True)
except (ImportError, AttributeError):
try:
import multiprocessing
cores = multiprocessing.cpu_count()
except (ImportError, NotImplementedError):
try:
import os
cores = os.cpu_count()
except (AttributeError, ImportError):
# Если ничего не сработало, используем безопасное значение
cores = 2

# Применяем коэффициент безопасности и округляем до целого числа
optimal = max(1, int(cores * safety_factor))

return optimal

# Примеры использования
cpu_workers = get_optimal_workers("cpu_bound")
io_workers = get_optimal_workers("io_bound", safety_factor=1.5) # Можно увеличить для IO-задач

print(f"Оптимальное количество рабочих процессов для CPU-задач: {cpu_workers}")
print(f"Оптимальное количество рабочих процессов для IO-задач: {io_workers}")

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

Практические рекомендации:

• Для CPU-интенсивных задач обычно оптимально использовать количество процессов, равное количеству физических ядер
• Для IO-интенсивных задач можно использовать количество процессов, превышающее количество логических ядер
• При работе с большими наборами данных учитывайте объем памяти: слишком много параллельных процессов могут привести к свопингу
• Всегда проводите бенчмаркинг для определения оптимального числа процессов/потоков для вашего конкретного приложения
• В производственных системах рассмотрите использование адаптивных алгоритмов, которые могут регулировать параллелизм в зависимости от текущей загрузки системы

Определение количества CPU — это фундамент эффективного системного программирования на Python. От выбора правильного метода зависит масштабируемость и производительность приложения. Простой модуль multiprocessing идеален для большинства случаев, os открывает доступ к низкоуровневым функциям, а psutil предоставляет всеобъемлющую картину системных ресурсов. Вооружившись этими инструментами, вы сможете создавать адаптивные решения, максимально эффективно использующие доступные ресурсы в любой среде выполнения.