Рабочие процессы

Рабочие процессы — это длительно работающие процессы, которые проверяют базу данных на наличие ожидающих задач, выполняют их и продвигают рабочие последовательности. Команда wp queuety work запускается внутри WP-CLI, а сама очередь считывается напрямую из MySQL с помощью наиболее подходящего доступного DB-драйвера Queuety.

Плагин также устанавливает резервный механизм WordPress cron, запускаемый раз в минуту, для сред без доступа к оболочке. На этой странице описаны выделенные рабочие процессы WP-CLI — рекомендуемый режим развертывания для более высокой пропускной способности.

Запуск рабочего процесса

wp queuety work

Это запускает один рабочий процесс в очереди default. Он работает в цикле: получить задачу, выполнить ее, повторить. Если очередь пуста, процесс ждет QUEUETY_WORKER_SLEEP секунду (по умолчанию: 1), а затем проверяет ее снова.

Параметры

# Process a specific queue
wp queuety work --queue=emails

# Process at most one job and exit
wp queuety work --once

# Run 4 concurrent workers
wp queuety work --workers=4

# Scale between 2 and 6 workers as backlog changes
wp queuety work --queue=providers --min-workers=2 --max-workers=6

Приоритеты нескольких очередей

Один рабочий процесс может обрабатывать несколько очередей в порядке приоритета. Передайте --queue список имен очередей, разделенных запятыми:

wp queuety work --queue=critical,default,low

Рабочий процесс пытается получить задачу из каждой очереди в указанном порядке. Сначала он проверяет critical. Если в critical есть задача, он обрабатывает ее. Если critical пуста, процесс переходит к default, а затем к low. Это означает, что очереди с более высоким приоритетом всегда обрабатываются раньше очередей с более низким приоритетом.

Как работает порядок приоритетов

На каждой итерации рабочий процесс пытается получить задачу из первой очереди в списке. Если очередь пуста или приостановлена, он переходит к следующей и продолжает, пока не найдет задачу или не проверит весь список. Если задач нигде нет, рабочий процесс ждет и повторяет попытку. При такой строгой модели приоритетов постоянный поток задач critical будет лишать ресурсов default и low. Если очередям с низким приоритетом нужна гарантированная пропускная способность, используйте выделенные рабочие процессы.

Использование нескольких очередей в PHP

Методы Worker::run() и Worker::flush() принимают строку или массив:

// Comma-separated string
Queuety::worker()->run( 'critical,default,low' );

// Array
Queuety::worker()->run( [ 'critical', 'default', 'low' ] );

// flush() also supports multi-queue
$count = Queuety::worker()->flush( 'critical,default,low' );

Стратегии развертывания

Выделенные рабочие процессы для каждой очереди дают полный контроль над параллелизмом и изоляцией. Для каждой очереди создается собственный process(es) рабочего процесса, поэтому другие очереди не смогут лишить ее ресурсов:

[program:queuety-critical]
command=wp queuety work --queue=critical
numprocs=4

[program:queuety-default]
command=wp queuety work --queue=default
numprocs=2

[program:queuety-low]
command=wp queuety work --queue=low
numprocs=1

Порядок приоритетов в одном рабочем процессе проще контролировать; этот вариант хорошо подходит, когда очереди с низким приоритетом могут ждать:

[program:queuety-worker]
command=wp queuety work --queue=critical,default,low
numprocs=2

Можно также объединить оба подхода: выделить рабочие процессы для critical, чтобы гарантировать пропускную способность, а для остальных очередей использовать рабочий процесс с приоритетным порядком:

[program:queuety-critical]
command=wp queuety work --queue=critical
numprocs=2

[program:queuety-fallback]
command=wp queuety work --queue=default,low
numprocs=2

Допуск с учетом ресурсов

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

Общее описание модели, в том числе ее связь с метаданными задач и бюджетами рабочих последовательностей, см. в разделе Управление ресурсами.

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

Недавние журналы выполнения также используются для составления простого профиля допуска. Когда допуск с учетом ресурсов включен, Queuety проверяет среднюю длительность и пиковое потребление памяти обработчика за последнее время перед запуском. Если текущий рабочий процесс уже близок к заданному пределу памяти, очередь или группа поставщиков достигла взвешенного бюджета, в текущем контейнере, похоже, не осталось достаточно памяти с учетом наблюдаемого профиля обработчика или запуск в стиле --once близок к пределу времени, Queuety откладывает задачу вместо начала работы, которая вряд ли завершится без ошибок.

Включите эту политику в определении задачи, а не в команде рабочего процесса:

class CallProviderStep implements \Queuety\Step {
    public function handle( array $state ): array {
        // ...
    }

    public function config(): array {
        return [
            'concurrency_group' => 'providers',
            'concurrency_limit' => 3,
            'cost_units' => 5,
        ];
    }
}

Те же метаданные можно назначить классам отправляемых задач через общедоступные свойства или отдельному вызову отправки через PendingJob::concurrency_group() и PendingJob::cost_units().

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

Жизненный цикл рабочего процесса

Рабочий процесс корректно останавливается после завершения текущей задачи при выполнении любого из следующих условий:

После остановки рабочий процесс завершает работу без ошибок. В рабочей среде диспетчер процессов автоматически запускает его снова.

Корректная обработка сигналов пока не поддерживается (#4). Отдельный процесс wp queuety work не устанавливает обработчик сигналов, поэтому SIGTERM или SIGINT (Ctrl+C) немедленно завершает его, даже посреди выполнения задачи. Такая задача остается в состоянии processing, пока механизм восстановления устаревших задач не сбросит ее через QUEUETY_STALE_TIMEOUT секунд, поэтому обработчики должны безопасно переносить повторный запуск. Пул рабочих процессов пересылает SIGTERM дочерним процессам, но те на него не реагируют, поэтому пул продолжает работать, пока они не остановятся по одной из указанных выше причин. Чтобы остановить пул, задайте диспетчеру процессов тайм-аут остановки, по истечении которого он отправит SIGKILL.

Параллельная работа нескольких процессов

Флаг --workers=N создает N дочерних процессов с помощью pcntl_fork(). Каждый дочерний процесс устанавливает собственное подключение к базе данных и обрабатывает задачи независимо. Родительский процесс следит за дочерними и перезапускает завершившиеся с ошибкой.

wp queuety work --queue=default --workers=4

Требования: Должны быть установлены расширения pcntl и posix PHP (без pcntl команда завершится с ошибкой «pcntl extension is required for --workers=N.»). Каждый дочерний процесс использует собственное подключение к базе данных, поэтому настройка MySQL max_connections должна учитывать пул. Для адаптивных пулов (--min-workers/--max-workers) действуют те же требования.

Адаптивные пулы рабочих процессов

Адаптивные пулы поддерживают минимальное количество готовых рабочих процессов, а затем увеличиваются или уменьшаются в заданном диапазоне:

wp queuety work --queue=providers --min-workers=2 --max-workers=6

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

Масштабирование по-прежнему выполняется осторожно. Когда Queuety может считывать показатели памяти контейнера или хоста, он отказывается добавлять дочерние процессы, если оставшейся памяти недостаточно для полного набора ресурсов еще одного рабочего процесса.

Цикл масштабирования на стороне родительского процесса настраивается с помощью следующих параметров:

Восстановление устаревших задач

Если рабочий процесс завершится с ошибкой во время обработки задачи, та останется в статусе processing. Длительно работающий процесс примерно раз в минуту проверяет, не находятся ли задачи в статусе processing дольше QUEUETY_STALE_TIMEOUT секунд. Для устаревшей задачи, у которой остались попытки, повторный запуск назначается после задержки между попытками; если все попытки исчерпаны, задача перемещается в архив, а ее рабочая последовательность помечается как завершившаяся с ошибкой. Рабочие процессы --once и резервный механизм WP-Cron эту проверку не выполняют.

Восстановите устаревшие задачи вручную:

wp queuety recover

Режим обработки очереди до конца

Обработать все ожидающие задачи и завершить работу. Удобно для тестирования или разовой пакетной обработки:

wp queuety flush
wp queuety flush --queue=emails

Развертывание в рабочей среде

В рабочей среде используйте диспетчер процессов, чтобы рабочие процессы продолжали работать. Ниже описаны два распространенных варианта.

Supervisord

[program:queuety-default]
command=wp queuety work --queue=default
directory=/var/www/html
autostart=true
autorestart=true
numprocs=2
process_name=%(program_name)s_%(process_num)02d
stdout_logfile=/var/log/queuety-default.log
stderr_logfile=/var/log/queuety-default-error.log

[program:queuety-emails]
command=wp queuety work --queue=emails
directory=/var/www/html
autostart=true
autorestart=true
numprocs=1
stdout_logfile=/var/log/queuety-emails.log
stderr_logfile=/var/log/queuety-emails-error.log

Systemd

[Unit]
Description=Queuety Worker (default queue)
After=mysql.service

[Service]
Type=simple
User=www-data
WorkingDirectory=/var/www/html
ExecStart=/usr/local/bin/wp queuety work --queue=default
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

Включите службу и запустите ее:

sudo systemctl enable queuety-worker
sudo systemctl start queuety-worker

Советы по развертыванию

Запускайте отдельные обработчики для отдельных очередей, если нужна более строгая изоляция. Используйте --workers=N для задач, ограниченных производительностью CPU, а при преимущественно операциях ввода-вывода отдавайте предпочтение обработчикам с индивидуальным управлением. Установите QUEUETY_WORKER_MAX_JOBS, если хотите периодически перезапускать процессы для освобождения памяти, и следите за количеством buried с помощью вебхуков или метрик.