摘要：基于有序集實(shí)現(xiàn)延遲任務(wù)執(zhí)行，比如某個(gè)時(shí)間給某個(gè)用戶發(fā)短信，訂單過期處理，等等我是在框架上寫的，實(shí)現(xiàn)起來很簡單對(duì)于一些不是很復(fù)雜的應(yīng)用足夠了，目前在公司項(xiàng)目中使用，后臺(tái)進(jìn)程并沒有實(shí)現(xiàn)多進(jìn)程，不多說，貼代碼，不回排版，見諒命令行腳本執(zhí)行方法這

基于redis有序集實(shí)現(xiàn)延遲任務(wù)執(zhí)行，比如某個(gè)時(shí)間給某個(gè)用戶發(fā)短信，訂單過期處理，等等
我是在tp5框架上寫的，實(shí)現(xiàn)起來很簡單,對(duì)于一些不是很復(fù)雜的應(yīng)用足夠了，目前在公司項(xiàng)目中使用，后臺(tái)進(jìn)程并沒有實(shí)現(xiàn)多進(jìn)程，
不多說，貼代碼，不回排版，見諒

1、命令行腳本 執(zhí)行方法:php think delay-queue queuename(這是有序集的key)

namespace appcommand;

use appcommonlibdelayqueueDelayQueue;
use thinkconsoleCommand;
use thinkconsoleInput;
use thinkconsoleOutput;
use thinkDb;

class DelayQueueWorker extends Command
{
    const COMMAND_ARGV_1 = "queue";

    protected function configure()
    {
        $this->setName("delay-queue")->setDescription("延遲隊(duì)列任務(wù)進(jìn)程");
        $this->addArgument(self::COMMAND_ARGV_1);
    }

    protected function execute(Input $input, Output $output)
    {
        $queue = $input->getArgument(self::COMMAND_ARGV_1);
        //參數(shù)1 延遲隊(duì)列表名,對(duì)應(yīng)與redis的有序集key名
        while (true) {
            DelayQueue::getInstance($queue)->perform();
            usleep(300000);
        }
    }
}

庫類目錄結(jié)構(gòu)

config.php 里是redis連接參數(shù)配置

RedisHandler.php只實(shí)現(xiàn)有序集的操作，重連機(jī)制還沒有實(shí)現(xiàn)

namespace appcommonlibdelayqueue;

class RedisHandler
{
    public $provider;
    private static $_instance = null;

    private function __construct() {
        $this->provider = new Redis();
        //host port
        $config = require_once "config.php";
        $this->provider->connect($config["redis_host"], $config["redis_port"]);
    }

    final private function __clone() {}

    public static function getInstance() {
        if(!self::$_instance) {
            self::$_instance = new RedisHandler();
        }
        return self::$_instance;
    }

    /**
     * @param string $key 有序集key
     * @param number $score 排序值
     * @param string $value 格式化的數(shù)據(jù)
     * @return int
     */
    public function zAdd($key, $score, $value)
    {
        return $this->provider->zAdd($key, $score, $value);
    }

    /**
     * 獲取有序集數(shù)據(jù)
     * @param $key
     * @param $start
     * @param $end
     * @param null $withscores
     * @return array
     */
    public function zRange($key, $start, $end, $withscores = null)
    {
        return $this->provider->zRange($key, $start, $end, $withscores);
    }

    /**
     * 刪除有序集數(shù)據(jù)
     * @param $key
     * @param $member
     * @return int
     */
    public function zRem($key,$member)
    {
        return $this->provider->zRem($key,$member);
    }

}

延遲隊(duì)列類

namespace appcommonlibdelayqueue;

class DelayQueue
{

    private $prefix = "delay_queue:";

    private $queue;

    private static $_instance = null;

    private function __construct($queue) {
        $this->queue = $queue;
    }

    final private function __clone() {}

    public static function getInstance($queue = "") {
        if(!self::$_instance) {
            self::$_instance = new DelayQueue($queue);
        }
        return self::$_instance;
    }

    /**
     * 添加任務(wù)信息到隊(duì)列
     *
     * demo DelayQueue::getInstance("test")->addTask(
     *    "appcommonlibdelayqueuejobTest",
     *    strtotime("2018-05-02 20:55:20"),
     *    ["abc"=>111]
     * );
     *
     * @param $jobClass
     * @param int $runTime 執(zhí)行時(shí)間
     * @param array $args
     */
    public function addTask($jobClass, $runTime, $args = null)
    {

        $key = $this->prefix.$this->queue;

        $params = [
            "class" => $jobClass,
            "args"  => $args,
            "runtime" => $runTime,
        ];

        RedisHandler::getInstance()->zAdd(
            $key,
            $runTime,
            serialize($params)
        );
    }

    /**
     * 執(zhí)行job
     * @return bool
     */
    public function perform()
    {
        $key = $this->prefix.$this->queue;
        //取出有序集第一個(gè)元素
        $result = RedisHandler::getInstance()->zRange($key, 0 ,0);

        if (!$result) {
            return false;
        }

        $jobInfo = unserialize($result[0]);

        print_r("job: ".$jobInfo["class"]." will run at: ". date("Y-m-d H:i:s",$jobInfo["runtime"]).PHP_EOL);

        $jobClass = $jobInfo["class"];

        if(!@class_exists($jobClass)) {
            print_r($jobClass." undefined". PHP_EOL);
            RedisHandler::getInstance()->zRem($key, $result[0]);
            return false;
        }

        // 到時(shí)間執(zhí)行
        if (time() >= $jobInfo["runtime"]) {
            $job = new $jobClass;
            $job->setPayload($jobInfo["args"]);
            $jobResult = $job->preform();
            if ($jobResult) {
                // 將任務(wù)移除
                RedisHandler::getInstance()->zRem($key, $result[0]);
                return true;
            }
        }

        return false;
    }

}

異步任務(wù)基類：

namespace appcommonlibdelayqueue;

class DelayJob
{

    protected $payload;

    public function preform ()
    {
        // todo
        return true;
    }


    public function setPayload($args = null)
    {
        $this->payload = $args;
    }

}

所有異步執(zhí)行的任務(wù)都卸載job目錄下,且要繼承DelayJob，你可以實(shí)現(xiàn)任何你想延遲執(zhí)行的任務(wù)

如:

namespace appcommonlibdelayqueuejob;

use appcommonlibdelayqueueDelayJob;

class Test extends DelayJob
{

    public function preform()
    {
        // payload 里應(yīng)該有處理任務(wù)所需的參數(shù)，通過DelayQueue的addTask傳入
        print_r("test job".PHP_EOL);
        return true;
    }

}

使用方法:

假設(shè)用戶創(chuàng)建了一個(gè)訂單，訂單在10分鐘后失效，那么在訂單創(chuàng)建后加入:

DelayQueue::getInstance("close_order")->addTask(
     "appcommonlibdelayqueuejobCloseOrder", // 自己實(shí)現(xiàn)的job
     strtotime("2018-05-02 20:55:20"), // 訂單失效時(shí)間
     ["order_id"=>123456] // 傳遞給job的參數(shù)
 );

close_order 是有序集的key

命令行啟動(dòng)進(jìn)程

php think delay-queue close_order