摘要：諸如此類，隊(duì)列的應(yīng)用范圍是如此之廣。方案抽象到更高一層，開發(fā)一套通用異步處理隊(duì)列適用于任何復(fù)雜的業(yè)務(wù)邏輯那么，作為架構(gòu)師，使用隊(duì)列的做法，將抽象層和業(yè)務(wù)層分離，可具有良好的擴(kuò)展性和可維護(hù)性。

一、隊(duì)列使用場景：為什么需要隊(duì)列
在web開發(fā)中，我們經(jīng)常會(huì)遇到需要處理批量任務(wù)的時(shí)候，這些批量任務(wù)可能是用戶提交的，也可能是當(dāng)系統(tǒng)被某個(gè)事件觸發(fā)時(shí)需要進(jìn)行批量處理的，面對(duì)這樣的任務(wù)，如果是用戶提交的批量任務(wù)，初級(jí)程序員只能讓用戶觸發(fā)提交動(dòng)作后，等待服務(wù)器處理完畢，并且將結(jié)果返回到瀏覽器，期間用戶不能關(guān)掉瀏覽器窗口，如果數(shù)據(jù)比較大，或者處理速度比較慢，那用戶體驗(yàn)將會(huì)因此受到直接影響。但是當(dāng)我們使用某訊或者某浪的郵箱時(shí)，點(diǎn)擊群發(fā)郵件之后，只需等待很短的時(shí)間，瀏覽器提示提交成功，正在發(fā)送之類的信息時(shí)，用戶就可以關(guān)掉瀏覽器，稍后，收件地址欄里的郵箱將陸續(xù)收到該群發(fā)郵件，再比如群發(fā)定時(shí)郵件，以及當(dāng)商城系統(tǒng)中有客戶下單，客戶，客服，倉庫等相關(guān)人員收到訂單郵件信息。諸如此類，隊(duì)列的應(yīng)用范圍是如此之廣。

二 :普通工程師的解決方案和架構(gòu)師的解決方案
方案1：建表存郵件，消息等，用定時(shí)程序取出發(fā)送。

方案2：抽象到更高一層，開發(fā)一套通用異步處理隊(duì)列適用于任何復(fù)雜的業(yè)務(wù)邏輯
那么，作為架構(gòu)師，使用隊(duì)列的做法，將抽象層和業(yè)務(wù)層分離，可具有良好的擴(kuò)展性和可維護(hù)性。相比較而言就高明了許多，下面就我們介紹一下自定義隊(duì)列的實(shí)現(xiàn)思路和方法。

三 :隊(duì)列總體設(shè)計(jì)

1：需要隊(duì)列程序，提供加入隊(duì)列接口和取隊(duì)列接口等
2：需要存儲(chǔ)隊(duì)列，文件或者數(shù)據(jù)庫
3：需要定時(shí)程序取出隊(duì)列并執(zhí)行
4：其它擴(kuò)展功能：優(yōu)先級(jí)，日志，定時(shí)等

代碼的目錄結(jié)構(gòu)如下，每個(gè)文件的作用用//注釋來標(biāo)明
|--addTask.php //添加任務(wù)到隊(duì)列的例子
|--cronMission.php //定時(shí)任務(wù)調(diào)度程序,例如linux中受crontab直接調(diào)用的文件，業(yè)務(wù)邏輯工程師可以在這個(gè)文件中靈活定義自己的隊(duì)列任務(wù)，從而不用每個(gè)隊(duì)列任務(wù)都需要上服務(wù)器修改crontab，從而在安全性，便捷性方面有很大提高
|--db.php //數(shù)據(jù)庫操作
|--db.sql //建立隊(duì)列需要用到的基本表結(jié)構(gòu)
|--doQueue.php //執(zhí)行隊(duì)列任務(wù)
|--Queue.class.php //隊(duì)列核心業(yè)務(wù)在這里定義，包括將任務(wù)加入隊(duì)列，讀隊(duì)列，更改隊(duì)列任務(wù)狀態(tài)
|--sendMsg.php //隊(duì)列要實(shí)現(xiàn)具體任務(wù)的業(yè)務(wù)接口，比如現(xiàn)有系統(tǒng)的發(fā)送消息的接口，這里例子中因?yàn)閷⒋岁?duì)列程序和現(xiàn)有系統(tǒng)系統(tǒng)集成，用寫入日志來演示

四 :隊(duì)列具體實(shí)現(xiàn)一：建任務(wù)存儲(chǔ)表
1：
先來個(gè)最基本的：

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：處理業(yè)務(wù)的接口文件
param：處理業(yè)務(wù)的接口文件需要接收的參數(shù)
status：任務(wù)處理狀態(tài)，0為未處理，處理完畢更改為1

五 、隊(duì)列具體實(shí)現(xiàn)二：定義調(diào)用接口

寫核心類Queue.class.php， 實(shí)現(xiàn)基本接口：
1：加入隊(duì)列接口
l //$param1 為執(zhí)行任務(wù)的程序，$param2 為程序參數(shù)，可以為序列化的數(shù)據(jù)
l $cqueue->add($param1,$param2);
2: 讀取隊(duì)列接口
l $tasks = $cqueue->getQueueTask($limit = 1000);
3：更新任務(wù)狀態(tài)
l $cqueue->updateTaskStatus($id);
4：a2s是自定義的一個(gè)數(shù)組轉(zhuǎn)換字符串方法，這里不要使用json_encode，容易出現(xiàn)問題，同樣，從數(shù)據(jù)庫中取出轉(zhuǎn)換為數(shù)組的時(shí)候，使用s2a方法
l $re = $cqueue->add("sendMsg.php", Queue::a2s($arr));

 4,//發(fā)信息的人的UID
     *      "uids" => array(6,234,34,67,7888,2355), //接收信息的人的UID
     *      "content" => "xxxxx",//信息內(nèi)容
     *  );
     * $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;
        }
    }


    /**
     * 讀取任務(wù)隊(duì)列
     * 
     * @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;
     }

    /**
     * 更新任務(wù)狀態(tài)
     * 
     * @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;
    }

}

?>

接下來就是部署調(diào)度了， 這個(gè)呢也不是就配置crontab那么簡單，要可維護(hù)性，可擴(kuò)展性好，能多進(jìn)程并發(fā)執(zhí)行，能監(jiān)控隊(duì)列運(yùn)行狀態(tài)等等，太長了，這個(gè)下一篇講。

本內(nèi)容來自本人講的公開課：http://www.ucai.cn/train?f=17