new_sdk_wu/www/new_sdk/application/crontab/controller/asyncCallback.php

<?php
namespace app\crontab\controller;

use GuzzleHttp\Client as Client;
use think\Db;
use think\Log;
use think\queue\Job;
use GuzzleHttp\Exception\GuzzleException;
use app\common\model\PayCpinfo;

/**
 * 游戏支付回调异步回调
 */
class asyncCallback
{
    private $data;
    
    /**
     * fire方法是消息队列默认调用的方法
     *  调用6次后失败则删除任务
     *  每次失败后延迟120秒再执行
     *
     * @param Job $job 当前的任务对象
     * @param array|mixed $data 发布任务时自定义的数据
     */
    public function fire(Job $job, $data)
    {

        Log::info("-------");
        Log::info("# redis队列开始了 orderid =".$data['orderid'].' attempts='.$job->attempts());
        
        //通过这个方法可以检查这个任务已经重试了几次了
        if ($job->attempts() > 5) {
            Log::info("# 重试次数已达上限 {$job->attempts()}");
            $job->delete();
            return;
        }
        
        // 如有必要,可以根据业务需求和数据库中的最新数据,判断该任务是否仍有必要执行.
        $isJobStillNeedToBeDone = $this->checkDatabaseToSeeIfJobNeedToBeDone($data);
        if ($isJobStillNeedToBeDone) {
            Log::info("# 发现通知已完成，队列 deleted");
            $job->delete();
            return;
        }
        
        $isJobDone = $this->doJob($data);
        if ($isJobDone) {
            //如果任务执行成功， 记得删除任务
            $job->delete();
            
            Log::info("# 订单处理完成，队列 deleted");
            //print("<info>Hello Job has been done and deleted" . "</info>\n");
        } else {
            if ($job->attempts() > 5) {
                //通过这个方法可以检查这个任务已经重试了几次了
                
                Log::info("# 重试次数已达上限,当前 {$job->attempts()} 次");
                //print("<warn>Hello Job has been retried more than 5 times!" . "</warn>\n");
                $job->delete();
                // 也可以重新发布这个任务
                //print("<info>Hello Job will be availabe again after 2s."."</info>\n");
                //$job->release(2); //$delay为延迟时间，表示该任务延迟2秒后再执行
            } else {
                Log::info("# 订单重试,当前 {$job->attempts()} 次");
                $job->release(120); //$delay为延迟时间，表示该任务延迟120秒后再执行
            }
        }
    }
    
    /**
     * 有些消息在到达消费者时,可能已经不再需要执行了
     * @param array|mixed $data 发布任务时自定义的数据
     * @return boolean                 任务执行的结果
     * @throws \think\db\exception\DataNotFoundException
     * @throws \think\db\exception\ModelNotFoundException
     * @throws \think\exception\DbException
     */
    private function checkDatabaseToSeeIfJobNeedToBeDone($data)
    {
        $payCpinfoModel = new PayCpinfo;
        
        //1 查询数据库判断是否已经更新
        $where['orderid'] = $data['orderid'];
        $where['status']  = 0;
        $where['payflag'] = 1;
        $this->data = $info = $payCpinfoModel->where($where)->find();
        if ($info) {
            return false;
        }
        return true;
    }
    
    /**
     * 根据消息中的数据进行实际的业务处理
     * @param array|mixed $data 发布任务时自定义的数据
     * @return boolean                 任务执行的结果
     */
    private function doJob($data)
    {
        // 将字符串参数转为数组形式
        if (!is_array($data['params'])) parse_str($data['params'], $data['params']);
        
        $guzzle = new Client();
        try {
            $result = $guzzle->request($data['http_method'], $data['url'], [
                'form_params' => $data['params'],
                'timeout'     => 30,
            ]);
            
        } catch (GuzzleException $exception) {
            Log::info("# doJob {$data['orderid']} 订单通知失败： " . $exception->getMessage());
            
            //todo 错误信息记录
            return false;
        }
        if (0 != strcasecmp($result->getBody(), $data['success_flag'])) {
            return false;
        }

        return $this->update();
    }
    
    private function update()
    {
        $payCpinfoModel = new PayCpinfo;
        $res = $payCpinfoModel->where(['id' => $this->data['id']])->update([
            'status'      => 1,
            'update_time' => time(),
        ]);
        if (!$res) {
            return false;
        }
        return true;
    }
}