vendor/codeages/biz-framework/src/Queue/Worker.php

这里只显示每个文件的开头 60 行。登录后可以解锁完整代码。

微信扫码关注公众号，代码整理好后第一时间通知你。

<?php

namespace Codeages\Biz\Framework\Queue;

use Codeages\Biz\Framework\Queue\Driver\Queue;
use Symfony\Component\Lock\LockInterface;
use Psr\Log\LoggerInterface;
use Symfony\Component\Lock\Exception\LockAcquiringException;
use Symfony\Component\Lock\Exception\LockConflictedException;

class Worker
{
    const EXIT_CODE_MEMORY_EXCEEDED = 1;

    const EXIT_CODE_TIMEOUT = 2;

    const EXIT_CODE_EXCEPTION = 3;

    const EXIT_CODE_RUNNING = 4;

    protected $queue;

    protected $options;

    protected $shouldQuit = false;

    protected $failer;

    protected $lock;

    protected $logger;

    public function __construct(Queue $queue, JobFailer $failer, LockInterface $lock, LoggerInterface $logger, array $options = array())
    {
        $this->queue = $queue;
        $this->failer = $failer;
        $this->lock = $lock;
        $this->logger = $logger;
        $this->options = array_merge(array(
            'memory_limit' => 256,
            'sleep' => 2,
            'tries' => 0,
            'once' => false,
            'stop_when_idle' => false,
        ), $options);
    }

    public function run()
    {
        try {
            $acquired = $this->lock->acquire();
        } catch (LockConflictedException $e) {
            $this->logger->error($this->createMessage("Acquire lock error: {$e->getMessage()}"));
        } catch (LockAcquiringException $e) {
            $this->logger->error($this->createMessage("Acquire lock error: {$e->getMessage()}"));
        }

        if (!$acquired) {
            $this->logger->warning($this->createMessage('Acquire lock failed, because other process is running.'));
            $this->stop(self::EXIT_CODE_RUNNING);