Redis實戰之實現定時執行任務

需求

• 異步執行任務
• 支持定時執行
• 支持取消任務
• 保障快速執行

技術背景

• 基於redis實現
• php

實現

基於redis的 sorted set + hash，實現定時執行任務的Demo

sorted set 介紹：

• redis有序集合，且不容許重複的成員，不一樣的是每一個元素都會關聯一個double類型的分數
• redis正是經過分數來爲集合中的成員進行從小到大的排序，有序集合的成員是惟一的,但分數(score)卻能夠重複

思路：

• 使用sortset類型，將member[成員] =【任務標識】，score[分數] =【定時時間戳】
• 使用hash類型，將【任務標識】對應的任務數據JSON存到hash中 key =【任務標識】，value =【任務數據JSON】
• 解決及時消耗，能夠運行多個進程進行並行執行

php實現代碼DEMO

<?php
class DoTest
{
    private const YIELD_KEY = 'yield:list';
    private const YIELD_DATA_KEY = 'yield:data';

    public function run()
    {
        $bbj = RedisClient::instance()->bbj();
        //獲取排序（低到高）中第一個task_id
        $data = $bbj->zRange(self::YIELD_KEY, 0, 0, true);
        if (empty($data)) {
            echo "無數據" . PHP_EOL;
            return null;
        }
        $mem = array_keys($data)[0];
        $ts = array_values($data)[0];
        $now = time();
        //校驗是否到時
        if ($ts > $now) {
            echo "還未到時間，無需操做" . PHP_EOL;;
            return null;
        }
        //移除集合，多進程並執行到時任務(只能被成功移除一次)
        $row = $bbj->zRem(self::YIELD_KEY, $mem);
        if (empty($row)) {
            echo "已經被剔除" . PHP_EOL;;
            return false;
        }
        //獲取當前要執行的任務數據JSON
        $dataJson = $bbj->hGet(self::YIELD_DATA_KEY, $mem);
        //todo 執行定時任務業務邏輯
        var_export($data);
        var_export($dataJson);
        //使用完後刪除任務數據JSON
        $bbj->hdel(self::YIELD_DATA_KEY, $mem);
        return true;
    }

    public function add(int $time, $i, string $content)
    {
        $data = [
            'msg' => $content . $time
        ];
        $bbj = RedisClient::instance()->bbj();
        $dataJson = json_encode($data);
        $taskId = $time . '_' . $i;
        $isSc = $bbj->zAdd(self::YIELD_KEY, $time, $taskId);
        if ($isSc) $isSc = $bbj->hSet(self::YIELD_DATA_KEY, $taskId, $dataJson);
        var_export($isSc);
    }

    public function addFeature($i)
    {
        $time = Carbon::now()->addMinute()->timestamp;
        $this->add($time, $i, '將來執行內容');
    }

    public function addCurrent($i)
    {
        $time = time();
        $this->add($time, $i, '立刻執行內容');
    }
    
    /**
     * 取消定時任務，根據任務ID
     * @param $taskId
     */
    public function removeYield($taskId)
    {
        $bbj = RedisClient::instance()->bbj();
        $bbj->zRem(self::YIELD_KEY, $taskId);
        $bbj->hDel(self::YIELD_DATA_KEY, $taskId);
    }
}

監控定時任務隊列

<?php
$dt = new DoTest();
while (true) {
    $rt = $dt->run();
    if (is_null($rt)) {
        sleep(1);
    }
}

添加當前執行和將來執行任務

<?php
$dt = new DoTest();
while (true) {
     for ($i = 0; $i < 100000; $i++) {
        // 添加當即執行當前任務
        $dt->addCurrent($i);
        // 添加待執行將來任務
        $dt->addFeature($i);
    }
}

取消定時任務

<?php
$dt = new DoTest();
$dt->removeYield('taskId');

解決場景

• 定時短信發送/email發送
• 定時執行??任務

---

首發於Github🌈大話WEB開發，歡迎Star 🥰