成人国产在线小视频_日韩寡妇人妻调教在线播放_色成人www永久在线观看_2018国产精品久久_亚洲欧美高清在线30p_亚洲少妇综合一区_黄色在线播放国产_亚洲另类技巧小说校园_国产主播xx日韩_a级毛片在线免费

資訊專欄INFORMATION COLUMN

基于Redis的任務(wù)調(diào)度設(shè)計方案

褰辯話 / 2433人閱讀

摘要:為了避免重復(fù)執(zhí)行的問題,我們需要引入一個有序集合存放正在執(zhí)行的任務(wù),命名為。最后以上講述了一個任務(wù)調(diào)度程序的逐步演變,設(shè)計方案很大程度上參考了。

原文鏈接:https://blog.breezelin.cn/scheme-redis-task-queue.html

一個網(wǎng)關(guān)服務(wù)器就跟快餐店一樣,總是希望客人來得快、去得也快,這樣在相同時間內(nèi)才可以服務(wù)更多的客人。如果快餐店的服務(wù)員在一個顧客點餐、等餐和結(jié)賬時都全程跟陪的話,那么這個服務(wù)員大部分時間都是在空閑的等待。應(yīng)該有專門的服務(wù)員負責(zé)點餐,專門的服務(wù)員負責(zé)送餐,專門的服務(wù)員負責(zé)結(jié)賬,這樣才能提高效率。同樣道理,網(wǎng)關(guān)服務(wù)器中也需要分工明確。舉個例子:

假設(shè)有一個申請發(fā)送重置密碼郵件的網(wǎng)關(guān)接口,須知道發(fā)送一封郵件可能會花費上好幾秒鐘,如果網(wǎng)關(guān)服務(wù)器直接在線上給用戶發(fā)送重置密碼郵件,高并發(fā)的情況下就很容易造成網(wǎng)絡(luò)擁擠。但實際上,網(wǎng)關(guān)服務(wù)器并非一定要等待郵件發(fā)送成功后才能響應(yīng)用戶,完全可以先告知用戶郵件會發(fā)送的,而后再在線下把郵件發(fā)送出去(就像快餐店里點餐的服務(wù)員跟顧客說先去找位置坐,飯菜做好后會有人給他送過去)。

那么是誰來把郵件發(fā)送出去呢?

任務(wù)隊列

為了網(wǎng)關(guān)接口能夠盡快響應(yīng)用戶請求,無需即時知道結(jié)果的耗時操作可以交由任務(wù)隊列機制來處理。
任務(wù)隊列機制中包含兩種角色,一個是任務(wù)生產(chǎn)者,一個是任務(wù)消費者,而任務(wù)隊列是兩者之間的紐帶:

生產(chǎn)者往隊列里放入任務(wù);

消費者從隊列里取出任務(wù)。

任務(wù)隊列的整體運行流程是:任務(wù)生產(chǎn)者把當(dāng)前操作的關(guān)鍵信息(后續(xù)可以根據(jù)這些信息還原出當(dāng)前操作)抽象出來,比如發(fā)送重置密碼的郵件,我們只需要當(dāng)前用戶郵箱和用戶名就可以了;任務(wù)生產(chǎn)者把任務(wù)放進隊列,實際就是把任務(wù)的關(guān)鍵信息存儲起來,這里會用到MySQL、Redis之類數(shù)據(jù)存儲工具,常用的是Redis;而任務(wù)消費者就不斷地從數(shù)據(jù)庫中取出任務(wù)信息,逐一執(zhí)行。

任務(wù)生產(chǎn)者的工作是任務(wù)分發(fā),一般由線上的網(wǎng)關(guān)服務(wù)程序執(zhí)行;任務(wù)消費者的工作是任務(wù)調(diào)度,一般由線下的程序執(zhí)行,這樣即使任務(wù)耗時再多,也不阻塞網(wǎng)關(guān)服務(wù)。

這里主要討論的是任務(wù)調(diào)度(任務(wù)消費者)的程序設(shè)計。

簡單直接

假設(shè)我們用Redis列表List存儲任務(wù)信息,列表鍵名是queues:default,任務(wù)發(fā)布就是往列表queues:default后追加數(shù)據(jù):


那么任務(wù)調(diào)度可以這樣簡單直接的實現(xiàn):

handle($task);
                continue;
            }
            sleep(1);
        }
    }

    public function handle($task) {
        // do something time-consuming
    }
}

$worker = new Worker;
$worker->schedule();
意外保險

上面代碼是直接從queues:default列表中移出第一個任務(wù)(lpop),因為handle($task)函數(shù)是一個耗時的操作,過程中若是遇到什么意外導(dǎo)致了整個程序退出,這個任務(wù)可能還沒執(zhí)行完成,可是任務(wù)信息已經(jīng)完全丟失了。保險起見,對schedule()函數(shù)進行以下修改:

handle($task);
                Redis::lpop("queues:default");
                continue;
            }
            sleep(1);
        }
    }
...

即在任務(wù)完成后才將任務(wù)信息從列表中移除。

延時執(zhí)行

queues:default列表中的任務(wù)都是需要即時執(zhí)行的,但是有些任務(wù)是需要間隔一段時間后或者在某個時間點上執(zhí)行,那么可以引入一個有序集合,命名為queues:default:delayed,來存放這些任務(wù)。任務(wù)發(fā)布時需要指明執(zhí)行的時間點$time


任務(wù)調(diào)度時,如果queues:default列表已經(jīng)空了,就從queues:default:delayed集合中取出到達執(zhí)行時間的任務(wù)放入queues:default列表中:

handle($task);
                Redis::lpop("queues:default");
                continue;
            }
            $seri_arr = Redis::zremrangebyscore("queues:default:delayed", 0, time());
            if($seri_arr) {
                Redis::rpush("queues:default", $seri_arr);
                continue;
            }
            sleep(1);
        }
    }
...
任務(wù)超時

預(yù)估任務(wù)正常執(zhí)行所需的最大時間值,若是任務(wù)執(zhí)行超過了這個時間,可能是過程中遇到一些意外,如果任由它繼續(xù)卡著,那么后面的任務(wù)就會無法被執(zhí)行了。
首先我們給任務(wù)設(shè)定一個時限屬性timeout,然后在執(zhí)行任務(wù)前先給進程本身設(shè)置一個鬧鐘信號,timeout后收到信號說明任務(wù)執(zhí)行超時,需要退出當(dāng)前進程(用supervisor守護進程時,進程自身退出,supervisor會自動再拉起)。
注意:pcntl_alarm($timeout)會覆蓋之前鬧鐘信號,而pcntl_alarm(0)會取消鬧鐘信號;任務(wù)超時后,當(dāng)前任務(wù)放入queues:default:delayed集合中延時執(zhí)行,以免再次阻塞隊列。

timeoutHanle($task);
                $this->handle($task);
                Redis::lpop("queues:default");
                continue;
            }
            $seri_arr = Redis::zremrangebyscore("queues:default:delayed", 0, time());
            if($seri_arr) {
                Redis::rpush("queues:default", $seri_arr);
                continue;
            }
            pcntl_alarm(0);
            sleep(1);
        }
    }

    public function timeoutHanle($task) {
        $timeout = (int)$task->timeout;
        if ($timeout > 0) {
            pcntl_signal(SIGALRM, function () {
                $seri = Redis::lpop("queues:default");
                Redis::zadd("queues:default:delayed", time()+10), $seri);
                posix_kill(getmypid(), SIGKILL);
            });
        }
        pcntl_alarm($timeout);
    }
...
并發(fā)執(zhí)行

上面代碼,直觀上沒什么問題,但是在多進程并發(fā)執(zhí)行的時候,有些任務(wù)可能會被重復(fù)執(zhí)行,是因為沒能及時將當(dāng)前執(zhí)行的任務(wù)從queues:default列表中移出,其他進程也可以讀取到。為了避免重復(fù)執(zhí)行的問題,我們需要引入一個有序集合SortedSet存放正在執(zhí)行的任務(wù),命名為queues:default:reserved。
首先任務(wù)是從queues:default列表中直接移出,然后開始執(zhí)行任務(wù)前先把任務(wù)放進queues:default:reserved集合中,任務(wù)完成了再從queues:default:reserved集合中移出。
再結(jié)合任務(wù)超時,假設(shè)一個任務(wù)執(zhí)行時間不可能超過60*60秒(可以按需調(diào)整),在queues:default列表為空的時候,queues:default:reserved集合中有任務(wù)已經(jīng)存放超過了60*60秒,那么有可能是某些進程在執(zhí)行任務(wù)是意外退出了,所以把這些任務(wù)放到queues:default:delayed集合中稍后執(zhí)行。

timeoutHanle($task);
                $this->handle($task);
                Redis::zrem("queues:default:reserved", $seri);
                continue;
            }
            $seri_arr = Redis::zremrangebyscore("queues:default:delayed", 0, time());
            if($seri_arr) {
                Redis::rpush("queues:default", $seri_arr);
                continue;
            }
            $seri_arr = Redis::zremrangebyscore("queues:default:reserved", 0, time()-60*60);
            if($seri_arr) {
                foreach($seri_arr as $seri) {
                    Redis::zadd("queues:default:delayed", time()+10, $seri);
                }
            }

            sleep(1);
        }
    }

    public function timeoutHanle($task) {
        $timeout = (int)$task->timeout;
        if ($timeout > 0) {
            pcntl_signal(SIGALRM, function () use ($task) {
                $seri = serialize($task);
                Redis::zrem("queues:default:reserved", $seri);
                Redis::zadd("queues:default:delayed", time()+10), $seri);
                posix_kill(getmypid(), SIGKILL);
            });
        }
        pcntl_alarm($timeout);
    }
...
其他 失敗重試

以上代碼沒有檢驗任務(wù)是否執(zhí)行成功,應(yīng)該有任務(wù)失敗的處理機制:比如給任務(wù)設(shè)定一個最多重試次數(shù)屬性retry_times,任務(wù)每執(zhí)行一次retry_times,任務(wù)執(zhí)行失敗時,若是retry_times等于0,則將任務(wù)放入queues:default:failed列表中不在執(zhí)行;否則放入放到queues:default:delayed集合中稍后執(zhí)行。

休眠時間

以上代碼是進程忙時連續(xù)執(zhí)行,閑時休眠一秒,可以按需調(diào)整優(yōu)化。

事件監(jiān)聽

若是需要在任務(wù)執(zhí)行成功或失敗時進行某些操作,可以給任務(wù)設(shè)定成功操作方法afterSucceeded()或失敗操作方法afterFailed(),在相應(yīng)的時候回調(diào)。

最后

以上講述了一個任務(wù)調(diào)度程序的逐步演變,設(shè)計方案很大程度上參考了Laravel Queue。
用工具,知其然,知其所以然。

文章版權(quán)歸作者所有,未經(jīng)允許請勿轉(zhuǎn)載,若此文章存在違規(guī)行為,您可以聯(lián)系管理員刪除。

轉(zhuǎn)載請注明本文地址:http://systransis.cn/yun/28333.html

相關(guān)文章

  • 分布式任務(wù)框架之celery

    摘要:架構(gòu)消息代理,作為臨時儲存任務(wù)的中間媒介,為提供了隊列服務(wù)。生產(chǎn)者將任務(wù)發(fā)送到,消費者再從獲取任務(wù)。如果使用,則有可能發(fā)生突然斷電之類的問題造成突然終止后的數(shù)據(jù)丟失等后果。任務(wù)調(diào)度器,負責(zé)調(diào)度并觸發(fā)定時周期任務(wù)。 架構(gòu) showImg(https://segmentfault.com/img/bVbmDXa?w=831&h=413); Broker 消息代理,作為臨時儲存任務(wù)的中間媒...

    fredshare 評論0 收藏0

發(fā)表評論

0條評論

最新活動
閱讀需要支付1元查看
<