Команды в Symfony 7: разбор на живом проекте

Команда импорта каталога в интернет-магазине на Symfony 7.1 работала три недели, пока поставщик не прислал выгрузку на 200 тысяч позиций вместо привычных пятнадцати. Команда съела всю память PHP-процесса, упала с Allowed memory size exhausted, а половина товаров осталась в неопределённом состоянии: часть обновилась, часть нет, откатить нечего. Разбираем на этой команде, где Symfony Console провоцирует ошибки под объёмом и как переписать так, чтобы она пережила рост данных.

Речь только про сам класс команды и её взаимодействие с Doctrine и вводом-выводом. Организацию очередей и Messenger не трогаем — это отдельная тема, здесь рассматриваем синхронный Command, который запускают из cron.

Что за команда и почему она ломалась

Команда читает CSV поставщика, для каждой строки ищет товар по артикулу, обновляет цену и остаток или создаёт новый. Первая версия выглядела предсказуемо: прочитать файл в массив, пройти циклом, на каждой итерации $entityManager->persist(), в конце один flush().

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

• Весь CSV грузился в память — file() или fgetcsv в цикле с накоплением в массив держали все строки одновременно.
• Doctrine копила managed-объекты — каждый найденный и созданный товар оставался в Identity Map до flush(). Двести тысяч объектов Product с их прайсами и остатками — это гигабайты.
• Один flush() в конце — если процесс падал на 150-тысячной строке, не сохранялось вообще ничего либо, в другой реализации, сохранялось всё до момента падения без контроля.

Ни одна из этих проблем не видна на тестовых данных. Все три вылезают ровно тогда, когда объём вырос, — то есть в проде, ночью, без вас.

Наивная версия — как было

// Symfony 7.1, схематично — так делать не надо
#[AsCommand(name: 'catalog:import')]
class CatalogImportCommand extends Command
{
    public function __construct(private EntityManagerInterface $em)
    {
        parent::__construct();
    }

    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $rows = array_map('str_getcsv', file('/data/catalog.csv'));
        foreach ($rows as $row) {
            $product = $this->em->getRepository(Product::class)
                ->findOneBy(['sku' => $row[0]]) ?? new Product();
            $product->setSku($row[0]);
            $product->setPrice((float) $row[1]);
            $this->em->persist($product);
        }
        $this->em->flush();
        return Command::SUCCESS;
    }
}

Три вещи, которые надо изменить

1. Читать файл потоком, а не целиком

Вместо file(), который возвращает массив всех строк, откройте дескриптор и читайте построчно через fgetcsv. В память попадает одна строка за раз.

// Чтение потоком — память не растёт с размером файла
$handle = fopen('/data/catalog.csv', 'r');
if ($handle === false) {
    $output->writeln('<error>Не удалось открыть файл</error>');
    return Command::FAILURE;
}

while (($row = fgetcsv($handle)) !== false) {
    // обработка одной строки
}

fclose($handle);

2. Сбрасывать Doctrine пачками и чистить Identity Map

Ключевой приём для батч-обработки в Doctrine — flush() и clear() каждые N объектов. flush() пишет накопленное в базу, clear() отсоединяет управляемые сущности, освобождая память. После clear() все прежние ссылки на сущности становятся недействительными — это граничный случай, о котором ниже.

$batchSize = 500;
$processed = 0;

while (($row = fgetcsv($handle)) !== false) {
    [$sku, $price] = $row;

    $product = $this->em->getRepository(Product::class)
        ->findOneBy(['sku' => $sku]) ?? new Product();
    $product->setSku($sku);
    $product->setPrice((float) $price);
    $this->em->persist($product);

    if ((++$processed % $batchSize) === 0) {
        $this->em->flush();
        $this->em->clear(); // освобождаем Identity Map
    }
}

$this->em->flush(); // хвост, который не добил до кратного batchSize
$this->em->clear();

3. Дать оператору обратную связь через SymfonyStyle и ProgressBar

Команда, которая молчит десять минут, неотличима от зависшей. SymfonyStyle и ProgressBar входят в компонент Console и не требуют отдельной установки.

use Symfony\Component\Console\Style\SymfonyStyle;

$io = new SymfonyStyle($input, $output);
$io->title('Импорт каталога');

// если общее число строк известно заранее — передайте его в progressBar
$progressBar = $io->createProgressBar();
$progressBar->start();

// внутри цикла:
$progressBar->advance();

// после цикла:
$progressBar->finish();
$io->success(sprintf('Обработано строк: %d', $processed));

Команда без прогресса и без возврата кода ошибки — это чёрный ящик, который cron запускает вслепую. Половина работы консольной команды — честно сообщить, что и как отработало.

Сравнение подходов к записи

Для каталога в 200 тысяч строк с бизнес-логикой на сущностях средний вариант — рабочий компромисс. Если строк миллионы и вам не нужны события Doctrine и валидация сущностей, дешевле обойти ORM и писать через Connection из DBAL пачками INSERT ... ON DUPLICATE KEY UPDATE — но тогда вы теряете lifecycle-события и отвечаете за целостность сами.

Где это ломается

Ссылки после clear() становятся мёртвыми. Если вы сохранили ссылку на сущность до clear() и обращаетесь к ней после — получите отсоединённый объект. Категории и другие связанные сущности, которые вы find() один раз до цикла, после первого clear() отвалятся. Либо не чистите их (получайте по id внутри пачки заново), либо используйте getReference().

findOneBy на каждой строке — это N запросов. Двести тысяч строк дают двести тысяч SELECT по одному. Даже с индексом на sku это узкое место. Если артикулы известны, читайте их пачкой и одним запросом WHERE sku IN (...) подтягивайте существующие товары в память пачки.

SQL-логгер копит запросы. В dev-окружении Doctrine и профайлер логируют каждый запрос в память. Запуская длинную команду, убедитесь, что работаете в APP_ENV=prod, иначе память утечёт даже с правильным clear.

Нет обработки битой строки. Одна строка с пустым артикулом или буквами в цене уронит всю команду. Оборачивайте обработку строки в try/catch, считайте ошибки и в конце выводите отчёт, а не падайте на первой аномалии.

Что померить и настроить дальше

После переписывания снимите два показателя на реальном объёме: пиковую память (memory_get_peak_usage(true) в конце команды) и время. На стабильной памяти пик не должен расти с числом строк — если растёт, где-то остался неочищенный накопитель.

• Подберите $batchSize замером: 500 — разумный старт, но оптимум зависит от размера сущности и настроек СУБД.
• Оберните пачку в транзакцию через Connection::transactional(), если нужна атомарность на уровне пачки.
• Для длинных ночных задач посмотрите в сторону вынесения импорта в Symfony Messenger — это позволит дробить работу на сообщения и переживать перезапуски воркера.
• Настройте таймаут и уведомление в cron: команда, зависшая на два часа, должна кого-то разбудить.

Точные сигнатуры configure(), атрибута #[AsCommand] и методов SymfonyStyle сверяйте с документацией вашей минорной версии Symfony — API компонента Console стабилен между минорами 7.x, но детали лучше проверить перед копированием.