Процессы

Процесс — это долгоживущий воркер: он поднимается один раз и работает, пока его не остановят. Слушает очередь, опрашивает оборудование, держит подключение к внешней системе. В отличие от запроса, у него нет конца — есть цикл.

Что такое процесс и зачем

Процесс — компонент приложения, который выполняется непрерывно, вне цикла «запрос — ответ».

Проблема. Часть работы не помещается в модель запроса. Слушать очередь надо постоянно, а не когда кто-то зашёл на страницу. Опрос оборудования по SNMP должен идти сам по себе. Демон, читающий из сокета, обязан держать подключение открытым неделями.

Обычное решение — отдельный PHP-скрипт под nohup или юнитом systemd. И тут же всплывает всё, о чём при написании скрипта не думают:

• по SIGTERM скрипт умирает посреди обработки задачи, теряя её;
• «остановить» — это найти PID, а «проверить, жив ли» — грепнуть ps;
• запустили дважды — работают оба, и никто об этом не узнает;
• у скрипта своя загрузка, свой конфиг и свои логи, отдельно от приложения.

Решение. Winter даёт стереотип: вы описываете что делать в цикле, а фреймворк берёт на себя остальное — сигналы и мягкую остановку, защиту от второго запуска, регистрацию в системе, отчёт о состоянии и общий с приложением контейнер.

Быстрый старт

Наследуйте Process и опишите тело в методе run():

<?php

namespace Main\Process;

use Flytachi\Winter\DI\Attribute\Autowired;
use Flytachi\Winter\Kernel\Process\Stereotype\Process;

class InboxWatcher extends Process
{
  #[Autowired] private DocumentImporter $importer;

  public function run(): void
  {
      $this->logger->info('watching /var/inbox');

      while ($this->isRunning()) {
          $files = glob('/var/inbox/*.xml');

          if ($files === []) {
              $this->sleep(2);          // пусто — подождём
              continue;
          }

          foreach ($files as $file) {
              $this->markBusy();
              $this->importer->import($file);
              unlink($file);
              $this->markIdle();
          }
      }
  }
}

Объявите его в манифесте приложения — и он поднимется вместе с остальным:

#[EnableWeb]
#[EnableProcess(\Main\Process\InboxWatcher::class)]
final class Application extends WinterApplication { /* ... */ }

Либо запускайте отдельно, без приложения:

php call process main.process.InboxWatcher start -d   # в фоне
php call process main.process.InboxWatcher status     # что с ним
php call process main.process.InboxWatcher stop       # мягко остановить

Тело процесса

Цикл и isRunning()

run() — единственный метод, который вы обязаны реализовать. Внутри он почти всегда содержит цикл, и условие цикла должно быть именно isRunning():

while ($this->isRunning()) {
  // ...
}

Метод возвращает false, как только пришёл сигнал остановки. Так остановка становится кооперативной: процесс не убивают на полуслове, он сам доходит до конца витка и выходит. Цикл while (true) лишает вас этой возможности.

Пауза — sleep()

Обычный sleep() из PHP блокирует процесс целиком: сигнал остановки будет ждать конца паузы, а под Swoole вместе с процессом встанут и все параллельные задачи. Используйте метод стереотипа:

$this->sleep(1.5);   // секунды, дробные допустимы

Он не блокирует рантайм и прерывается остановкой: если сигнал пришёл во время паузы, вместо досыпания будет брошено InterruptedException.

try {
  $this->sleep(30);
} catch (InterruptedException) {
  // остановка во время паузы — вернуть задачу в очередь и выйти
}

Ловить его не обязательно: если исключение не поймать, процесс просто мягко свернётся, а блоки finally отработают.

Единицы работы — markBusy() и markIdle()

Между этими вызовами процесс считается занятым. Это даёт две вещи:

1. Занятый процесс не прерывают посреди работы — остановка дождётся markIdle().
2. Занятость видна в call process status и учитывается супервизором демона.

$this->markBusy();
try {
  $this->handle($job);
} finally {
  $this->markIdle();
}

Признак жизни — touch()

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

Внутри такой работы подавайте признак жизни сами:

foreach ($rows as $i => $row) {
  $this->recalculate($row);          // считаем, ничего не ждём

  if ($i % 1000 === 0) {
      $this->touch();                // «я жив, просто занят»
  }
}

Вызов дешёвый и защищён от частых повторов, так что ставить его в цикле безопасно.

Что ещё доступно внутри

Стереотип отдаёт телу процесса небольшой набор инструментов. Всё перечисленное доступно через $this без каких-либо объявлений.

Два из них стоит пояснить отдельно.

$this->logger объявлять не нужно. Он уже есть и уже назван по классу процесса, поэтому #[Autowired] private LoggerInterface $logger в процессе — лишняя строка:

$this->logger->info('batch processed', ['count' => count($batch)]);

activity() отвечает на вопрос «я сейчас занят?» — с учётом и явных markBusy(), и незавершённых задач из spawn(). Пригождается, когда решение брать ли новую порцию зависит от того, разгребли ли предыдущую:

use Flytachi\Winter\Kernel\Process\Activity;

while ($this->isRunning()) {
  if ($this->activity() === Activity::BUSY) {
      $this->sleep(0.2);        // предыдущая пачка ещё в работе
      continue;
  }

  foreach ($this->jobs->takeBatch(50) as $job) {
      $this->spawn(fn() => $this->handle($job));
  }
}

Без этой проверки цикл продолжал бы набирать задачи, пока не упрётся в $concurrency, и накопил бы очередь ожидающих корутин.

requestStop() останавливает процесс по собственному решению — то же самое, что внешний сигнал. Уместен, когда работа закончилась по существу:

if ($this->migration->isComplete()) {
  $this->logger->info('migration finished, stopping');
  $this->requestStop();
}

Параллельная работа — spawn()

Когда задачи независимы, их можно обрабатывать одновременно:

while ($this->isRunning()) {
  foreach ($this->jobs->takeBatch(10) as $job) {
      $this->spawn(fn() => $this->handle($job));
  }

  $this->sleep(1);
}

Под Swoole каждая задача идёт в своей корутине; пока одна ждёт базу или сеть, работают остальные. Процесс считается занятым, пока есть незавершённые задачи, и не завершится, не дождавшись их.

Ограничить число одновременных задач — свойством класса:

protected int $concurrency = 20;   // 0 (по умолчанию) — без ограничения

Результат задачи

spawn() возвращает Future — тот же объект-обещание, что и у #[Async]. Если результат не нужен, возврат просто отбрасывают, как в примере выше. Если нужен — соберите обещания и заберите значения:

$futures = [];

foreach ($this->sources->all() as $source) {
  $futures[$source->id] = $this->spawn(fn() => $this->fetch($source));
}

foreach ($futures as $id => $future) {
  $this->results->save($id, $future->get());   // ждёт, если ещё не готово
}

Здесь все обращения уходят разом, а get() дожидается каждого по очереди — общее время равно самому долгому из них, а не их сумме.

Исключение внутри задачи не роняет процесс — оно сохраняется в Future и всплывает при get(). Задача, чей результат никто не забирает, свою ошибку никому не покажет, поэтому в таких местах ловите её сами и пишите в лог.

Настройки

Три защищённых свойства управляют поведением процесса.

class InboxWatcher extends Process
{
  protected int $concurrency = 20;
  protected float $grace = 30.0;
  protected ?string $processTitle = 'inbox';
}

$grace стоит задать осознанно: при значении по умолчанию остановка ждёт завершения работы неограниченно долго. Это правильно, когда задачу нельзя бросить, и опасно при развёртывании — зависший воркер задержит выкатку. Значение в десятки секунд обычно разумный компромисс.

Имя процесса в ps

Процесс называет себя сам, и по умолчанию берёт короткое имя класса. Если в системе удобнее видеть другое имя — задайте свойство:

protected ?string $processTitle = 'inbox-watcher';

winter-proc: inbox-watcher

Когда одного свойства мало — например, имя должно включать параметр запуска, — переопределите titleName(). Метод возвращает имя без префикса рантайма, префикс подставляется поверх.

Реакция на сигналы

Все хуки необязательны — переопределяйте только нужные.

protected function onTerminate(): void
{
  $this->logger->info('stopping, will finish the current job');
}

protected function onReload(): void
{
  $this->settings->reload();     // SIGHUP — перечитали и работаем дальше
}

protected function onShutdown(): void
{
  $this->connection->close();    // выполнится в любом случае
}

onShutdown() — место для освобождения ресурсов: он отрабатывает на всех путях выхода, включая фатальную ошибку. Остановка сама по себе гарантирована и без onTerminate() — этот хук нужен, только если вы хотите на неё отреагировать.

Остановить себя изнутри можно вызовом requestStop() — это тот же эффект, что от SIGTERM, но по собственному решению.

Три способа запуска

Процесс — обычный класс, и поднять его можно тремя путями. Разница не в поведении самого процесса, а в том, кто его запускает и кто за ним следит.

Управление из кода

Те же операции, что делает консоль, доступны статическими методами прямо на классе процесса. Это и есть третий способ: запустить обработку по нажатию кнопки в админке, остановить зависший импорт, показать состояние на странице.

Запуск и остановка

use Main\Process\InboxWatcher;

$pid = InboxWatcher::dispatch();          // ушёл в фон, вернулся PID

InboxWatcher::stop();                     // мягкая остановка: true, если сигнал ушёл

Аргумент dispatch() задаёт, куда девать вывод дочернего процесса. По умолчанию он отбрасывается; передайте путь, если нужно сохранить:

InboxWatcher::dispatch('/var/log/inbox-watcher.out');

Повторный dispatch() при уже работающем процессе бросит ProcessAlreadyRunningException — это и есть способ не запустить второй экземпляр.

Чтение состояния

status() возвращает null, если процесс не работает, — это самая простая проверка «жив ли»:

if (InboxWatcher::status() === null) {
  InboxWatcher::dispatch();
}

Когда нужны подробности, у объекта состояния доступны поля:

Плюс метод getStartedAt(): string — то же время в читаемом виде.

Состояния, которые может вернуть $state:

В обычной жизни процесса встречаются первые три: status() возвращает null, как только процесса нет, поэтому TERMINATED и FAILED вы увидите не у него, а у воркеров флота — там запись переживает сам процесс.

Расход ресурсов не собирается без запроса, потому что стоит вызова системной утилиты. Запрашивайте его осознанно:

$status = InboxWatcher::status(usage: true);

$status->usage?->cpu;      // процент процессора
$status->usage?->rssKb;    // занятая память, КБ
$status->usage?->elapsed;  // сколько работает

Пример: управление из админки

<?php

namespace Main\Admin;

use Flytachi\Winter\Kernel\Http\Response\ResponseEntity;
use Flytachi\Winter\Kernel\Http\Stereotype\Controller;
use Flytachi\Winter\Kernel\Process\ProcessAlreadyRunningException;
use Flytachi\Winter\Kernel\Route\Annotation\{GetMapping, PostMapping, RequestMapping};
use Main\Process\InboxWatcher;

#[RequestMapping('admin/inbox-watcher')]
class ProcessController extends Controller
{
  #[GetMapping]
  public function show(): ResponseEntity
  {
      $status = InboxWatcher::status(usage: true);

      return $status === null
          ? ResponseEntity::ok(['running' => false])
          : ResponseEntity::ok($status);
  }

  #[PostMapping('start')]
  public function start(): ResponseEntity
  {
      try {
          return ResponseEntity::ok(['pid' => InboxWatcher::dispatch()]);
      } catch (ProcessAlreadyRunningException $e) {
          return ResponseEntity::conflict(['message' => $e->getMessage()]);
      }
  }

  #[PostMapping('stop')]
  public function stop(): ResponseEntity
  {
      return InboxWatcher::stop()
          ? ResponseEntity::accepted(['stopping' => true])
          : ResponseEntity::notFound(['message' => 'Процесс не запущен']);
  }
}

Остановка возвращает 202, а не 200, намеренно: stop() только посылает сигнал, а процесс завершится, когда доработает текущую единицу работы.

Запрет на совместную работу

Иногда два процесса не должны работать одновременно — например, обработчик очереди и разовая миграция, которые трогают одни и те же записи. Проверка «не запускать второй экземпляр себя» вынесена в отдельный метод, и его можно расширить:

use Main\Process\Migration;

class OutboxWorker extends Process
{
  protected static function ensureNotRunning(): void
  {
      parent::ensureNotRunning();      // не запускать второй экземпляр себя
      Migration::ensureNotRunning();   // ...и не запускать, пока идёт миграция
  }
}

Метод вызывается один раз перед запуском, в start() и dispatch(). Если любая из проверок не прошла, будет ProcessAlreadyRunningException и процесс просто не поднимется.

Управление из консоли

php call process list                      # все процессы проекта и их состояние
php call process main.process.Worker       # запустить в текущем терминале
php call process main.process.Worker start -d   # запустить в фоне
php call process main.process.Worker stop       # мягкая остановка (SIGTERM)
php call process main.process.Worker status     # состояние
php call process main.process.Worker status -v  # плюс расход CPU и памяти

Класс указывается точечной нотацией: main.process.Worker разворачивается в Main\Process\Worker. У команды есть короткий псевдоним — call proc.

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

Process ● RUNNING
PID          48213
State        RUNNING
Activity     busy
Started      2026-08-11 14:02:31 +03:00
Uptime       2h 14m
Concurrency  20

Примеры

Опрос внешнего источника

Простейшая форма: разбудились, сходили, поспали.

<?php

namespace Main\Process;

use Flytachi\Winter\DI\Attribute\Autowired;
use Flytachi\Winter\Kernel\Process\Stereotype\Process;

class RatesPoller extends Process
{
  #[Autowired] private RateGateway $gateway;
  #[Autowired] private RateRepository $rates;

  protected ?string $processTitle = 'rates';
  protected float $grace = 10.0;

  public function run(): void
  {
      while ($this->isRunning()) {
          $this->markBusy();

          try {
              $this->rates->saveAll($this->gateway->fetch());
          } catch (\Throwable $e) {
              $this->logger->error('rates poll failed', ['error' => $e->getMessage()]);
          } finally {
              $this->markIdle();
          }

          $this->sleep(60);
      }
  }
}

Обратите внимание на try/catch вокруг работы: исключение, выпущенное из run(), завершит процесс. Для регулярной работы ошибку почти всегда нужно поймать, записать и продолжить.

Сторож доступности

Процесс не обязан ничего обрабатывать — он может просто наблюдать. Здесь он опрашивает список сервисов и заводит тревогу, когда кто-то перестал отвечать.

<?php

namespace Main\Process;

use Flytachi\Winter\DI\Attribute\Autowired;
use Flytachi\Winter\Kernel\Process\InterruptedException;
use Flytachi\Winter\Kernel\Process\Stereotype\Process;

class HealthChecker extends Process
{
  #[Autowired] private ServiceRegistry $registry;
  #[Autowired] private HttpClient $http;
  #[Autowired] private Alerts $alerts;

  protected ?string $processTitle = 'health';

  public function run(): void
  {
      while ($this->isRunning()) {
          $this->markBusy();

          foreach ($this->registry->all() as $service) {
              $ok = $this->http->ping($service->url, timeout: 3);

              if (!$ok && $service->wasHealthy()) {
                  $this->alerts->fire("{$service->name} недоступен");
              }

              $this->registry->record($service, $ok);
          }

          $this->markIdle();

          try {
              $this->sleep(15);
          } catch (InterruptedException) {
              return;                      // остановили во время паузы
          }
      }
  }
}

Обработчик очереди

Классический случай: процесс разбирает накопленные задания.

<?php

namespace Main\Process;

use Flytachi\Winter\DI\Attribute\Autowired;
use Flytachi\Winter\Kernel\Process\Stereotype\Process;

class OutboxWorker extends Process
{
  #[Autowired] private OutboxRepository $outbox;
  #[Autowired] private Mailer $mailer;

  protected float $grace = 30.0;

  public function run(): void
  {
      while ($this->isRunning()) {
          $letter = $this->outbox->takeNext();

          if ($letter === null) {
              $this->sleep(2);
              continue;
          }

          $this->markBusy();

          try {
              $this->mailer->send($letter);
              $this->outbox->markSent($letter->id);
          } catch (\Throwable $e) {
              $this->outbox->release($letter->id);   // вернуть в очередь
              $this->logger->error('send failed', ['id' => $letter->id]);
          } finally {
              $this->markIdle();
          }
      }
  }
}

Ключевая деталь — release() в catch: задание, взятое из очереди, должно вернуться в неё при любом сбое, иначе оно потеряется вместе с процессом.

Параллельная обработка с ограничением

Когда работа упирается в ожидание сети, а не в процессор.

<?php

namespace Main\Process;

use Flytachi\Winter\DI\Attribute\Autowired;
use Flytachi\Winter\Kernel\Process\Stereotype\Process;

class WebhookSender extends Process
{
  #[Autowired] private WebhookRepository $hooks;
  #[Autowired] private HttpClient $http;

  protected int $concurrency = 50;    // не больше 50 запросов одновременно
  protected float $grace = 60.0;      // дать доставить начатое

  public function run(): void
  {
      while ($this->isRunning()) {
          $batch = $this->hooks->takeBatch(100);

          if ($batch === []) {
              $this->sleep(1);
              continue;
          }

          foreach ($batch as $hook) {
              $this->spawn(function () use ($hook) {
                  $this->http->post($hook->url, $hook->payload);
                  $this->hooks->markDelivered($hook->id);
              });
          }
      }
  }
}

Предел $concurrency здесь обязателен: без него пачка в сто тысяч записей попытается уйти одновременно. А $grace в минуту даёт начатым доставкам завершиться при остановке.

Освобождение ресурсов при перезапуске

Если процесс работает воркером демона, он создаётся форком — и наследует подключения родителя, которыми пользоваться нельзя.

protected function afterFork(): void
{
  parent::afterFork();          // обязательно: сбрасывает пулы фреймворка
  $this->rabbit->reconnect();   // и своё
}

Вызов parent::afterFork() первым — не формальность: именно он переоткрывает подключения к базе. У процесса, запущенного самостоятельно, этот хук не вызывается — форка нет.

Дальше

• Демоны — несколько воркеров одной логики под супервизией
• Состав приложения — где объявляется #[EnableProcess]
• Планировщик — если работа привязана ко времени, а не к очереди
• Асинхронные вызовы — параллельность внутри одной задачи
• Внедрение зависимостей — как процесс получает сервисы