停止通过 cron 作业运行的并发 php 作业作业

Question

我正在使用 CakePhp 的 Queuesadilla 作业队列插件来管理几个繁重的任务，例如从 excel 中提取数据等。我使用带有间隔的 cron 作业运行 shell 命令。但是，如果队列中列出了多个作业； shell 命令启动下一个工作的工作人员，而无需等待前一个工作的结束或失败。由于我想以 5 秒的非常小的间隔运行 shell 命令，并且某些作业需要 3 分钟才能完全运行；我希望工人等待上一份工作完成，直到那时才开始。

但是，当我使用 bin/cake CustomQueuesadilla 运行 cronjob 时；显然它从一个新的 PID 开始，依此类推，不知道前一个是否仍在运行。当我将工作人员的运行时间设置为 20 秒并将 cron 作业间隔设置为 3 分钟时；我有一份工作每 3 分钟工作一次。但这也意味着如果没有作业，它将等待 3 分钟 - 20 秒才能再次运行。我想减少这个时间窗口。

我也对架构变化和不同的库持开放态度。

Answer 1

队列与 Cron

您要实现的是循环队列。 Cron 仅用于预定事件。

1. 构建您的队列管理器/工作者
   • 这个工人calls the other commands。
   • 此类调用应包装在 try/catch 中，并检查退出代码。
   • 您可能应该记录每次通话的(time, command [, error])。
   • 将其构建为单例。
2. 添加一个 cron 条目以调用获取管理器（单例）的命令。
   • 作为单例模式的结果，如果它停止了，它将重新开始。
   • 除非您将队列的状态存储在该管理器（文件或数据库）之外，否则它可能总是在 cmd4 处死掉，并且永远不会运行后续命令。

前经理

class CommandQueueLoop {
    const QUEUE = [
        'app:command1' => [
            'command' => 'app:command1',
        ],
        'app:command2' => [
            'command' => 'app:command2',
            'somearg' => 123
        ],
        'app:command3' => [
            'command' => 'app:command3',
        ],
    ];

    private static $queueManager;

    private $activeQueue;
    private $app;
    private $input;
    private $output;
    private $running;

    public function __construct(Application $app, InputInterface $input, OutputInterface $output)
    {
        $this->app = $app;
        $this->input = $input;
        $this->output = $output;
        $this->running = false;
        $this->start();
    }

    public static function getInstance(Application $app, InputInterface $input, OutputInterface $output)
    {
        if (!self::$queueManager instanceof self) {
            self::$queueManager = new self();
        }
    }

    public function start(array $queue = null)
    {
        $this->activeQueue = $queue ?? self::QUEUE;
        $this->running = true;
        $this->run($this->activeQueue);
    }

    public function stop()
    {
        $this->running = false;
        $this->activeQueue = null;
    }

    private function run(array $queue)
    {
        if (!$this->running) {
            return;
        }

        foreach ($queue as $cmd => $args) {
            $this->call($cmd, $args);
        }

        $this->run($this->activeQueue);
    }

    private function call(string $command, array $args)
    {
        try {
            $cmd = $this->app->find($command);
            $inp = new ArrayInput($args);
            $returnCode = $cmd->run($inp, $this->output);
        } catch (Exception $e) {
            // log failure
            unset($this->activeQueue[$cmd]); // remove from queue
        }

        // check returnCode & log
    }
}

确保队列正在运行的 Cron 命令

添加这个命令，并且只有这个命令到 cron，每小时一次左右。当它运行时，如果命令已停止，它将重新启动。

class QueueLoopCommand extends Command
{
    ...

    public function execute(InputInterface $input, OutputInterface $output)
    {
        $queueLoop = CommandQueueLoop::getInstance($this->getApplication(), $input, $output);
    }
}

这里是龙

这个“管理器”需要非常仔细地编写，具有良好的日志记录和某种紧急停止功能。此外，该 Cron 命令应该调用“管理器”中的许多其他方法来检查状态、上次运行、仔细检查多个线程等，并对意外状态做一些事情。像这样的构建工具可以快速深入，并且可以像你不会相信的那样把你带入兔子洞。

主要危险：

• 内存泄漏
• 多个冲突线程
• 意外的比赛条件