seaBlueQueue/Jobs/Controller/CAddJob.Class.php

<?php
namespace Jobs\Controller;

use Mall\Framework\Factory;
use Mall\Framework\Core\ErrorCode;

use Jobs\Controller\CJobObject;
use Jobs\Controller\BaseController;
use Jobs\Model\MQueue;
use think\Error;

class CAddJob extends BaseController
{
    private $objMQueue;

    public function __construct()
    {
        parent::__construct();

        $config = Factory::config()->all();
        $this->config      = $config;
        $this->logger      = Factory::logs($this->config['logPath'] ?:'', $this->config['logSaveFileApp'] ?:'', $this->config['app_name'] ?:'');

        $this->objMQueue = MQueue::getQueue($this->config['job']['queue'], $this->logger);
        if(!is_object($this->objMQueue)){
            exit('初始化队列失败');
        }
        $this->objMQueue->setTopics($this->config['job']['topics']);
    }

    /**
     * 统一的添加队列任务入口
     */
    public function add()
    {

        /*$params['topicName'] = 'MyJob';                       // 推送到那个队列主题中
        $params['topicClass'] = 'Jobs\Model\MTopic\Log\MLog'; // 往队列中添加得任务脚本类文件
        $params['topicMethon'] = 'push';                     // 需要触发的类文件中的入口函数
        $params['topicMethonParams'] = [
            'userCenterId' => 1,
            'enterpriseId' => 15,
            'createTime'   => time(),//操作时间
            'no' => 'Y12455f4545', // 来源:
            'actionType'=> '新增订单', // 操作类型：登录/注册
            'operationData' => [
                "goodsId"=>12,
                "name"=>"商品名称"
            ], //本次操作请求的数据(可空) json格式
        ];    // 携带过去的参数*/

        //接收参数
        $params['topicName'] = $this->request->param('topicName');
        $params['topicClass'] = $this->request->param('topicClass');
        $params['topicMethon'] = $this->request->param('topicMethon');
        $params['topicMethonParams'] = $this->request->param('topicMethonParams');

        foreach ($params as $key => $value){
            if(empty($value)){
                parent::sendOutput('参数'.$key.'值为空', ErrorCode::$paramError);
            }
        }

        $jobExtras = [                              // 指定在队列中的执行优先级(redis版本队列不支持,带上就行)
            'delay' => mt_rand(0, 100) * 1000,
            'priority' => 1,
        ];

        if(!is_array($params['topicMethonParams'])){
            $params['topicMethonParams'] = json_decode(htmlspecialchars_decode($params['topicMethonParams'], ENT_QUOTES), true);
        }

        $job    = new CJobObject($params['topicName'], $params['topicClass'], $params['topicMethon'], $params['topicMethonParams'], $jobExtras);
        $result = $this->objMQueue->push($params['topicName'], $job);
        if($result) {
            parent::sendOutput('添加队列成功');
        }
        parent::sendOutput('添加队列失败',ErrorCode::$redisWriteError);
    }

}