PHP+Redis implementiert eine Verzögerungswarteschlange-PHP-Tutorial-php.cn

PHP+Redis implementiert eine Verzögerungswarteschlange

不言

Freigeben： 2023-03-25 13:18:01

Original

11377 Leute haben es durchsucht

Dieser Artikel stellt hauptsächlich die Implementierung einer verzögerten Warteschlange mit PHP+Redis vor, die einen gewissen Referenzwert hat. Jetzt kann ich ihn mit allen teilen, die ihn brauchen.

Implementierung einer verzögerten Aufgabenausführung basierend auf Redis Bestellsatz, z. B. das Senden von Textnachrichten an einen bestimmten Benutzer zu einem bestimmten Zeitpunkt, die Abwicklung des Bestellablaufs usw.
Ich habe es im tp5-Framework geschrieben. Es ist sehr einfach zu implementieren. Es reicht für einige nicht sehr komplexe Anwendungen . Es wird derzeit in den Projekten des Unternehmens verwendet, der Hintergrundprozess implementiert keinen Multiprozess.
Ich werde nicht viel sagen, ich werde nur den Code veröffentlichen und nicht auf den Satz zurückkommen, sorry

1. Befehlszeilenskript-Ausführungsmethode: PHP Think Delay-Queue Queuename (dies ist der ordnungsgemäße Set-Schlüssel)

namespace app\command;

use app\common\lib\delayqueue\DelayQueue;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\Db;

class DelayQueueWorker extends Command
{
    const COMMAND_ARGV_1 = &#39;queue&#39;;

    protected function configure()
    {
        $this->setName(&#39;delay-queue&#39;)->setDescription(&#39;延迟队列任务进程&#39;);
        $this->addArgument(self::COMMAND_ARGV_1);
    }

    protected function execute(Input $input, Output $output)
    {
        $queue = $input->getArgument(self::COMMAND_ARGV_1);
        //参数1 延迟队列表名,对应与redis的有序集key名
        while (true) {
            DelayQueue::getInstance($queue)->perform();
            usleep(300000);
        }
    }
}

Nach dem Login kopieren

Bibliotheksverzeichnisstruktur
PHP+Redis implementiert eine Verzögerungswarteschlange

config.php ist die Redis-Verbindungsparameterkonfiguration

RedisHandler implementiert nur geordnete Set-Operationen und der Wiederverbindungsmechanismus wurde noch nicht implementiert

namespace app\common\lib\delayqueue;

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

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

    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 格式化的数据
     * @return int
     */
    public function zAdd($key, $score, $value)
    {
        return $this->provider->zAdd($key, $score, $value);
    }

    /**
     * 获取有序集数据
     * @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);
    }

    /**
     * 删除有序集数据
     * @param $key
     * @param $member
     * @return int
     */
    public function zRem($key,$member)
    {
        return $this->provider->zRem($key,$member);
    }

}

Nach dem Login kopieren

Verzögerungswarteschlangenklasse

namespace app\common\lib\delayqueue;

class DelayQueue
{

    private $prefix = &#39;delay_queue:&#39;;

    private $queue;

    private static $_instance = null;

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

    final private function __clone() {}

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

    /**
     * 添加任务信息到队列
     *
     * demo DelayQueue::getInstance(&#39;test&#39;)->addTask(
     *    &#39;app\common\lib\delayqueue\job\Test&#39;,
     *    strtotime(&#39;2018-05-02 20:55:20&#39;),
     *    [&#39;abc&#39;=>111]
     * );
     *
     * @param $jobClass
     * @param int $runTime 执行时间
     * @param array $args
     */
    public function addTask($jobClass, $runTime, $args = null)
    {

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

        $params = [
            &#39;class&#39; => $jobClass,
            &#39;args&#39;  => $args,
            &#39;runtime&#39; => $runTime,
        ];

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

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

        if (!$result) {
            return false;
        }

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

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

        $jobClass = $jobInfo[&#39;class&#39;];

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

        // 到时间执行
        if (time() >= $jobInfo[&#39;runtime&#39;]) {
            $job = new $jobClass;
            $job->setPayload($jobInfo[&#39;args&#39;]);
            $jobResult = $job->preform();
            if ($jobResult) {
                // 将任务移除
                RedisHandler::getInstance()->zRem($key, $result[0]);
                return true;
            }
        }

        return false;
    }

}

Nach dem Login kopieren

Asynchrone Aufgabenbasisklasse:

namespace app\common\lib\delayqueue;

class DelayJob
{

    protected $payload;

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


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

}

Nach dem Login kopieren

Alle asynchron ausgeführten Aufgaben Deinstallieren Sie das Jobverzeichnis und erben Sie DelayJob. Sie können jede Aufgabe implementieren, deren Ausführung Sie verzögern möchten

wie zum Beispiel:

namespace app\common\lib\delayqueue\job;

use app\common\lib\delayqueue\DelayJob;

class Test extends DelayJob
{

    public function preform()
    {
        // payload 里应该有处理任务所需的参数，通过DelayQueue的addTask传入
        print_r(&#39;test job&#39;.PHP_EOL);
        return true;
    }

}

Nach dem Login kopieren

Verwendung:

Angenommen, der Benutzer erstellt eine Bestellung, die Bestellung läuft nach 10 Minuten ab, und fügen Sie sie dann hinzu, nachdem die Bestellung erstellt wurde:

DelayQueue::getInstance(&#39;close_order&#39;)->addTask(
     &#39;app\common\lib\delayqueue\job\CloseOrder&#39;, // 自己实现的job
     strtotime(&#39;2018-05-02 20:55:20&#39;), // 订单失效时间
     [&#39;order_id&#39;=>123456] // 传递给job的参数
 );

Nach dem Login kopieren

close_order ist der Schlüssel von bestellter Satz

Befehlszeilen-Startvorgang

PHP Think Delay-Queue Close_Order

Verwandte Empfehlungen:

PHP+Redis zur Implementierung der Sitzungsfreigabe

Das obige ist der detaillierte Inhalt vonPHP+Redis implementiert eine Verzögerungswarteschlange. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!