xyzshop/helper/task/manager.php

<?php

namespace task;

use Exception;
use trans_wapper;
use Log;

class manager
{
    private $mHandler;
    public function __construct()
    {
        $this->mHandler = new handler();
    }

    public function add_task($method, $params, $is_show = 1, $del_timeout = -1, $title = '', $mchid = 0)
    {
        if(empty($method) || empty($params)) {
            return false;
        }

        //task 唯一性
        $hash_fun = function ($method,$params) {
            $val = $method . '-' . serialize($params);
            return md5($val);
        };
        $task_hash = $hash_fun($method,$params);

        if(empty($title)) {
            $title_fun = "{$method}_title";
            $title = $this->mHandler->$title_fun($params);
        }

        $mod_task = Model('task');
        $item = $mod_task->getByHash($task_hash);

        if(!empty($item) && $del_timeout > 0)
        {
            if(time() - $item['add_time'] > $del_timeout) {
                $mod_task->where(['task_id' => $item['task_id']])->delete();
                $item = [];
            }
        }

        if(empty($item))
        {
            $item = ['type' => $method, 'title' => $title,
                     'params' => serialize($params), 'task_hash' => $task_hash,
                     'add_time' => time(), 'is_show' => $is_show, 'mchid' => $mchid];
            $task_id = $mod_task->insert($item);
            $item = $mod_task->where(['task_id' => $task_id])->master(true)->find();
        }

        return task_wrapper::CreateTask($item);
    }

    //弹出一个任务，并且将任务设为进行中
    public function pop_task($task_id,$merchant = false)
    {
        $mod_task = Model('task');
        $trans = new trans_wapper($mod_task,'pop_task');

        try
        {
            if($merchant)
            {
                $item = $mod_task->field('*')
                    ->where(['task_id' => ['gt',$task_id],'state' => task_wrapper::Waiting,'mchid' => ['gt',0]])
                    ->master(true)->lock(true)->find();
            } else {
                $item = $mod_task->field('*')
                    ->where(['task_id' => ['gt',$task_id],'state' => task_wrapper::Waiting,'mchid' => 0])
                    ->master(true)->lock(true)->find();
            }

            if(!empty($item)) {
                $mod_task->where(['task_id' => $item['task_id']])->update(['state' => task_wrapper::Processing,'act_time' => time()]);
            }
            $trans->commit();

            return $item;
        }
        catch (Exception $ex)
        {
            $trans->rollback();
            Log::record($ex->getMessage(),Log::ERR);
            return [];
        }
    }

    //任务结束，并且
    public function fini_task($task_id,$succ)
    {

    }

    public function del_task()
    {

    }

    public function getResult($task)
    {

    }

    public function handle($task)
    {
        $method = $task['type'];
        $params = unserialize($task['params']);

        $handler = new handler();
        $handler->set_task_id($task['task_id']);
        [$succ,$result] = $handler->$method($params);

        $mod_task = Model('task');
        if(!empty($result))
        {
            $mod_task->where(['task_id' => $task['task_id']])->update(['state' => task_wrapper::Handled,
                'result_state' => $succ ? 1 : 0, 'result' => serialize($result),
                'finish_time' => time()]);
        }

        return $task['task_id'];
    }
}