本身實現異步執行任務的隊列(一)

1、隊列使用場景:爲何須要隊列
在web開發中,咱們常常會遇到須要處理批量任務的時候,這些批量任務多是用戶提交的,也多是當系統被某個事件觸發時須要進行批量處理的,面對這樣的任務,若是是用戶提交的批量任務,初級程序員只能讓用戶觸發提交動做後,等待服務器處理完畢,而且將結果返回到瀏覽器,期間用戶不能關掉瀏覽器窗口,若是數據比較大,或者處理速度比較慢,那用戶體驗將會所以受到直接影響。可是當咱們使用某訊或者某浪的郵箱時,點擊羣發郵件以後,只需等待很短的時間,瀏覽器提示提交成功,正在發送之類的信息時,用戶就能夠關掉瀏覽器,稍後,收件地址欄裏的郵箱將陸續收到該羣發郵件,再好比羣發定時郵件,以及當商城系統中有客戶下單,客戶,客服,倉庫等相關人員收到訂單郵件信息。諸如此類,隊列的應用範圍是如此之廣。php

二 :普通工程師的解決方案和架構師的解決方案
方案1:建表存郵件,消息等,用定時程序取出發送。mysql

方案2:抽象到更高一層,開發一套通用異步處理隊列適用於任何複雜的業務邏輯
那麼,做爲架構師,使用隊列的作法,將抽象層和業務層分離,可具備良好的擴展性和可維護性。相比較而言就高明瞭許多,下面就咱們介紹一下自定義隊列的實現思路和方法。linux

三 :隊列整體設計程序員

1:須要隊列程序,提供加入隊列接口和取隊列接口等
2:須要存儲隊列,文件或者數據庫
3:須要定時程序取出隊列並執行
4:其它擴展功能:優先級,日誌,定時等web

代碼的目錄結構以下,每一個文件的做用用//註釋來標明
|--addTask.php //添加任務到隊列的例子
|--cronMission.php //定時任務調度程序,例如linux中受crontab直接調用的文件,業務邏輯工程師能夠在這個文件中靈活定義本身的隊列任務,從而不用每一個隊列任務都須要上服務器修改crontab,從而在安全性,便捷性方面有很大提升
|--db.php //數據庫操做
|--db.sql //創建隊列須要用到的基本表結構
|--doQueue.php //執行隊列任務
|--Queue.class.php //隊列核心業務在這裏定義,包括將任務加入隊列,讀隊列,更改隊列任務狀態
|--sendMsg.php //隊列要實現具體任務的業務接口,好比現有系統的發送消息的接口,這裏例子中由於將此隊列程序和現有系統系統集成,用寫入日誌來演示sql

四 :隊列具體實現一:建任務存儲表
1:
先來個最基本的:數據庫

CREATE TABLE`queue` (
 id int(11) NOT NULL auto_increment primarykey,
 taskphp varchar(128) NOT NULL default '',
 param text not null default '',
 status tinyint not null default 0,
 ctime timestamp NOT NULL default CURRENT_TIMESTAMP,
 KEY (ctime)
) ENGINE=InnoDBDEFAULT CHARSET=utf8;

字段解釋:
taskphp:處理業務的接口文件
param:處理業務的接口文件須要接收的參數
status:任務處理狀態,0爲未處理,處理完畢更改成1json

五 、隊列具體實現二:定義調用接口數組

寫核心類Queue.class.php, 實現基本接口:
1:加入隊列接口
l //$param1 爲執行任務的程序,$param2 爲程序參數,能夠爲序列化的數據
l $cqueue->add($param1,$param2);
2: 讀取隊列接口
l $tasks = $cqueue->getQueueTask($limit = 1000);
3:更新任務狀態
l $cqueue->updateTaskStatus($id);
4:a2s是自定義的一個數組轉換字符串方法,這裏不要使用json_encode,容易出現問題,一樣,從數據庫中取出轉換爲數組的時候,使用s2a方法
l $re = $cqueue->add("sendMsg.php", Queue::a2s($arr));瀏覽器

<?php
/**
 * 優才網公開課示例代碼
 *
 * 任務隊列實現
 *
 * @author 優才網全棧工程師教研組  張友林
 * @see http://www.ucai.cn
 */

include_once('db.php');
class Queue
{

    /**
     * 把任務扔到隊列
     *
     * @param string $taskphp   執行任務的程序
     * @param string $param     執行任務程序所用的參數

     * 例如,羣發消息加入隊列:
     * $arr = array(
     *      "uid" => 4,//發信息的人的UID
     *      "uids" => array(6,234,34,67,7888,2355), //接收信息的人的UID
     *      "content" => 'xxxxx',//信息內容
     *  );
     * $cqueue = new Queue();
     * $cqueue->add("/app/send_msg.php", serialize($arr));
     *
     */
    public function add($taskphp,$param)
    {
        $taskphp = mysql_real_escape_string($taskphp);
        //$param = mysql_real_escape_string($param);
        $param = $param;
        $sql = "insert into queue (taskphp, param) values('".$taskphp."', '".$param."')";
        $re = execute($sql);
        if ($re)
        {
            $pid = mysql_insert_id();
            return $pid;
        }
        else
        {
            return false;
        }
    }


    /**
     * 讀取任務隊列
     * 
     * @param string $limit 一次取多少條
     */
     public function getQueueTask($limit = 1000)
     {
        $limit = (int)$limit;
        $sql = "select id, taskphp, param from queue  where status = 0 order by id asc";
        $re = query($sql);
        return $re;
     }

    /**
     * 更新任務狀態
     * 
     * @param string $limit 一次取多少條
     */
     public function updateTaskByID($id)
     {
        $id = (int)$id;
        $mtime = time();
        $sql = "update queue  set status =1, mtime = ".$mtime." where id = ".$id;
        $re = execute($sql);
        return $re;
     }


     public static function a2s($arr)
    {
        $str = "";
        foreach ($arr as $key => $value)
        {
            if (is_array($value))
            {
                foreach ($value as $value2)
                {
                    $str .= urlencode($key) . "[]=" . urlencode($value2) . "&";
                }
            }
            else
            {
                $str .= urlencode($key) . "=" . urlencode($value) . "&";
            }
        }
        return $str;
    }

    public static function s2a($str)
    {
        $arr = array();
        parse_str($str, $arr);
        return $arr;
    }

}

?>

接下來就是部署調度了, 這個呢也不是就配置crontab那麼簡單,要可維護性,可擴展性好,能多進程併發執行,能監控隊列運行狀態等等,太長了,這個下一篇講。

本內容來自本人講的公開課:http://www.ucai.cn/train?f=17

相關文章
相關標籤/搜索