PHP-RESQUE - 實現(xiàn)重試

因為PHP-Resque 的重試部分需要自己寫，網(wǎng)上又沒啥輪子，而且resque也已經(jīng)很久不更新了，所以自己研究下resque的源碼，然后也借鑒了Laravel的隊列重試機制，實現(xiàn)了PHP-Resque的重試機制。

Resque地址

https://github.com/chrisboulton/php-resque

設(shè)計思路

1.這里需要閱讀resque源碼，resque的worker把失敗的隊列的數(shù)據(jù)都放在了"resque:failed"列表,數(shù)據(jù)如下。

2.我的項目是在yii2下統(tǒng)一處理resque的，打算開一個守護進程不斷重試這些失敗的隊列。
3.我直接resque源碼在新增了retry方法，在每一個failJob的payload增加了attempts嘗試次數(shù)，
一開始創(chuàng)建的時候attempts為0，重試之后attempts每次加1然后進行retry，直到到達嘗試數(shù)次才停止重試。

核心代碼

1. 調(diào)用重試的代碼

 /**
     * Retry Job
     */
    public function actionRetry()
    {
        $redis = new Redis('redis');
        while (true) {
            $json = $redis->rPop('resque:failed');
            $array = json_decode($json, true);
            if ($array) {
                $jobArray = $array['payload'];
                if ($jobArray['attempts'] < $this->retryTimes) {
                    //retry
                    \Resque::retry($jobArray, $array['queue']);
                    echo "Queued job " . $jobArray['id'] . ' has retry!' . "\n";
                } else {
                    //stop retry
                    $redis->lPush('resque:failed', [$json]);
                }
            }
            //take a sleep
            echo "*** Sleeping for ".$this->sleep. "\n";
            sleep($this->sleep);
        }

        return true;
    }

這里可以弄成守護進程一直處理失敗的隊列任務(wù)。

2. /php-resque/lib/Resque.php

/**
     * retry job and save it to the specified queue.
     *
     * @param array $jobArray The attempts of the the job .
     * array(
     *  'id'=> The id of the job
     *  'class' => The name of the class that contains the code to execute the job.
     *  'args' => Any optional arguments that should be passed when the job is executed.''
     *  'attempts'=> The retry attempts of the job
     * )
     * @param string $queue The name of the queue to place the job in.
     *
     * @return boolean
     */
    public static function retry($jobArray,$queue)
    {
        require_once dirname(__FILE__) . '/Resque/Job.php';
        $result = Resque_Job::retry($jobArray,$queue);
        if ($result) {
            Resque_Event::trigger('afterEnqueue', array(
                'class' => $jobArray['class'],
                'args'  => $jobArray['args'],
                'queue' => $queue,
            ));
        }

        return true;
    }

3. php-resque/lib/Resque/Job.php

/**
     * retry  job and save it to the specified queue.
     * *
     * @param array $jobArray The data of the job.
     * array(
     *  'id'=> The id of the job
     *  'class' => The name of the class that contains the code to execute the job.
     *  'args' => Any optional arguments that should be passed when the job is executed.''
     *  'attempts'=> The retry attempts of the job
     * )
     * @param string $queue The name of the queue to place the job in.
     *
     * @return string
     */
    public static function retry($jobArray,$queue)
    {
        $args = $jobArray['args'];
        if($args !== null && !is_array($args)) {
            throw new InvalidArgumentException(
                'Supplied $args must be an array.'
            );
        }
        $jobArray['attempts']++;
        Resque::push($queue, array(
            'class' => $jobArray['class'],
            'args'  => $args,
            'id'    => $jobArray['id'],
            'attempts'=>$jobArray['attempts']
        ));
        return true;
    }

結(jié)果：

我這里我設(shè)置了重試次數(shù)為3次，每隔5秒處理一個隊列。

注意

    1. 只有任務(wù)JOB實現(xiàn)類出現(xiàn)異常才會被重新扔回到"resque:failed"隊列里面。
    2. 對于正常的業(yè)務(wù)錯誤不需要重試，目前只考慮到對curl會出現(xiàn)的異常進行重試。