Laravel Horizon 定时 kill 掉旧的消费者进程

问题

生产环境使用了 horizon 来处理队列数据，但是在 supervisor 自动重启的过程中，发现有时候会有一些消费者进程在部署的时候无法正常被 kill 掉，导致一些旧的代码消费队列数据，从而引起报错。

原因是可能在重启的时候，那个消费者恰好在处理长时间的任务，导致无法被 kill 掉。但无法肯定是这个原因。

解决方法

每分钟 kill 掉那些不是当前 horizon 子孙进程的消费者（S 状态才会 kill 掉，保证正常业务不受影响）。

解决思路

1. 通过 shell_exec 来执行 ps、pstree 等命令查看当前的进程信息，从而找出那些异常的消费者进程 id。
2. 判断这些异常进程是否没有在处理任务。
3. kill 掉 S 状态的异常消费者进程。

注意：shell_exec 里面使用 ps aux 可能无法获取正常在终端运行的结果，需要使用 ps -efww 才可以获取完整的输出。

实现代码

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Illuminate\Support\Facades\Log;

class ClearInvalidQueueConsumer extends Command
{
    protected $signature = 'clear-invalid-queue-consumer';

    protected $description = '清除无效的队列消费者进程';

    public function handle()
    {
        $noParentPids        = $this->noParentPids();
        $sleepingProcessPids = $this->getSleepingProcessPids($noParentPids);

        if ($sleepingProcessPids) {
            Log::info('kill 掉无效的队列进程：', $sleepingProcessPids);
            $this->killInvalidProcess($sleepingProcessPids);
        }
    }

    public function noParentPids(): array
    {
        $pids = $this->getValidPids();

        $allPids = $this->getAllQueueProcessPids();

        return array_diff($allPids, $pids);
    }

    /**
     * kill 掉无效的进程
     *
     * @param array $pids
     */
    private function killInvalidProcess(array $pids)
    {
        foreach ($pids as $pid) {
            if (!posix_kill($pid, SIGTERM)) {
                Log::error("kill ${pid} error!");
            }

            if ($error = posix_get_last_error()) {
                Log::error('kill process error: ' . posix_strerror($error));
            }
//            system("kill -9 {$pid}");
        }
    }

    /**
     * 从 pid 数组里面筛选出睡眠状态的进程
     *
     * @param array $pids
     *
     * @return array
     */
    public function getSleepingProcessPids(array $pids): array
    {
        $result = [];

        foreach ($pids as $pid) {
            if ($this->isSleeping($pid)) {
                $result[] = $pid;
            }
        }

        return $result;
    }

    /**
     * 进程是否是睡眠状态
     *
     * @param $pid
     *
     * @return bool
     */
    public function isSleeping($pid): bool
    {
        $res = $this->grepPidStatus($pid);

        if ($res) {
            $status = $this->statusOfLine($res);
            if ($status == 'S') {
                return true;
            }
        }

        return false;
    }

    public function grepPidStatus($pid)
    {
        return shell_exec("ps -efww | grep php | grep ${pid} | grep -v grep");
    }

    /**
     * 获取当前所有的队列进程
     *
     * @return array
     */
    public function getAllQueueProcessPids(): array
    {
        $res     = $this->horizonProcesses();
        $allPids = [];

        foreach (explode("\n", $res) as $line) {
            if (!$line) {
                continue;
            }

            $allPids[] = $this->pidOfLine($line);
        }

        return $allPids;
    }

    public function horizonProcesses()
    {
        return shell_exec('ps -efww | grep horizon | grep -v grep');
    }

    /**
     * 当前有效的进程 id 数组
     *
     * @return array
     */
    public function getValidPids(): array
    {
        $horizonPid = $this->horizonProcessPid();
        if (!$horizonPid) {
            return [];
        }

        $res = $this->pstree($horizonPid);

        preg_match_all("/php,\d+/", $res, $matches);

        $pids = [];
        if ($matches) {
            foreach ($matches[0] as $match) {
                $pids[] = substr($match, 4);
            }
        }

        return $pids;
    }

    public function pstree($horizonPid)
    {
        return shell_exec("pstree -a {$horizonPid} -p");
    }

    /**
     * 获取 horizon 启动进程 id
     *
     * @return int|mixed
     */
    public function horizonProcessPid()
    {
        $res = $this->horizonPidLine();

        if ($res) {
            return $this->pidOfLine($res);
        }

        return 0;
    }

    public function horizonPidLine()
    {
        return shell_exec("ps -efww | grep -v grep | grep '/usr/lnmp/php/bin/php /home/www/api/default/current/artisan horizon'");
    }

    /**
     * 从 ps aux 的输出行里面获取进程状态 S -> 睡眠
     *
     * @param string $line
     *
     * @return mixed
     */
    public function statusOfLine(string $line)
    {
        return $this->fields($line)[7];
    }

    /**
     * 从 ps aux 的输出行里面获取 pid
     *
     * @param string $line
     *
     * @return mixed
     */
    public function pidOfLine(string $line)
    {
        return $this->fields($line)[1];
    }

    /**
     * ps aux 输出行里面空格分隔的单词列表
     *
     * @param string $line
     *
     * @return array
     */
    public function fields(string $line): array
    {
        $line = trim($line);

        $fields = explode(' ', $line);

        return array_values(array_filter($fields));
    }
}