摘要：方案介紹該方案出來的場景一天有一個業(yè)務需求，需要把我方的一些信息或訂單狀態(tài)等異步發(fā)起請求同步給第三方，這里就會出現(xiàn)定時時間和延遲時間消息的處理，考慮過很多消息隊列方案如云消息服務等。

該方案出來的場景：一天有一個業(yè)務需求，需要把我方的一些信息或訂單狀態(tài)等異步發(fā)起請求同步給第三方，這里就會出現(xiàn)定時時間和延遲時間消息的處理，考慮過很多消息隊列方案（如:rabbitmq、云消息服務等）。
不過最后公司定了因為該業(yè)務流量很小，不用做那么麻煩。所以就直接出了這個方案
該方案在50條消息/s，應該壓力不大，量大了就會出現(xiàn)一個消息延遲問題，如果不注重這個業(yè)務時間準確性，該方案承載的秒級處理在1000內(nèi)應該問題也不大

CREATE TABLE `open_queue` (
  `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
  `type` tinyint(4) unsigned NOT NULL DEFAULT "0" COMMENT "任務類型，可用于后續(xù)讀取不同任務失敗次數(shù)區(qū)分",
  `order_id` int(10) unsigned NOT NULL DEFAULT "0" COMMENT "業(yè)務綁定的訂單id等，根據(jù)自己自行設(shè)計業(yè)務id",
  `event_identity` varchar(30) NOT NULL DEFAULT "" COMMENT "事件表示分類，可不用，同type",
  `data` text NOT NULL COMMENT "需要處理的數(shù)據(jù)，json格式化",
  `try` tinyint(4) unsigned NOT NULL DEFAULT "1" COMMENT "特定任務需要當時就處理的次數(shù)，而不是發(fā)起回調(diào)請求的任務",
  `again` tinyint(1) unsigned NOT NULL DEFAULT "0" COMMENT "0不是重試記錄 1重試記錄，失敗后發(fā)起任務未1，否者未0",
  `status` tinyint(1) unsigned NOT NULL DEFAULT "0" COMMENT "是否被執(zhí)行 1是 0否",
  `create_time` int(10) unsigned NOT NULL DEFAULT "0" COMMENT "任務創(chuàng)建的時間戳",
  `at_time` int(11) unsigned NOT NULL DEFAULT "0" COMMENT "任務時間的時間戳",
  `task_status` tinyint(1) NOT NULL DEFAULT "0" COMMENT "執(zhí)行結(jié)果 -1重復消息系統(tǒng)主動取消 0未執(zhí)行 1執(zhí)行成功 2執(zhí)行失敗",
  `task_time` int(10) unsigned NOT NULL DEFAULT "0" COMMENT "執(zhí)行時間",
  PRIMARY KEY (`id`),
  KEY `IDX_EVENT` (`event_identity`),
  KEY `IDX_AT_TIME` (`at_time`) USING BTREE,
  KEY `IDX_ORDER` (`order_id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

下方代碼都是偽代碼，大家自行根據(jù)思路自己設(shè)計業(yè)務

該步驟可以省略，因為我方當時redis是單機部署，也沒做持久化,

$redisTypeLock; //lock類型
if (!IS_WIN) {
    $file = fopen($path . "/lock/{$redisTypeLock}.lock","w+");
    if (!flock($file, LOCK_EX | LOCK_NB)) {
        flock($file,LOCK_UN);
        fclose($file);
        exit;
    }
}

//成功獲得鎖 開始業(yè)務執(zhí)行

$count = get_redis_lLen("msgevent:" . $redisType);
if (!$count) {
    //無需要入數(shù)據(jù)庫訂單隊列，直接返回
    return false;
}

$successCount = 0;
for ($i = 0; $i < $loopLen; $i++) {
    $data = get_redis_lPop("msgevent:" . $redisType);
    //把redis數(shù)據(jù)格式化存入到數(shù)據(jù)庫持久化
    $result = $this->saveToMysql($data);
    if (!$result) {
        //如果執(zhí)行失敗，重新推入redis隊列
        get_redis_lPush("msgevent:" . $redisType, $data);
    }

    $successCount++;
}

if (!IS_WIN) {
    flock($file, LOCK_UN);
    fclose($file);
}

這個入口是定時秒級處理腳本和處理小于當前時間的腳本調(diào)用的入口

public function task_run()
{
    $timeType = $_GET["tt"] ? : "now";
    //獲取當前秒需要處理的所有消息, 依次路由后執(zhí)行
    //如果你redis無需存入mysql，你可以redis的list結(jié)構(gòu)實現(xiàn), 一次全部取出數(shù)據(jù)集合，并刪除list
    //如果接下去業(yè)務里發(fā)現(xiàn)發(fā)起請求失敗了，重新把任務分配給redis建立恢復list，list名為t+需要什么時間執(zhí)行的時間戳
    $lists = $this->getTaskLists($timeType);
    foreach ($lists as $item) {
        $this->task("event:" . $item["event_type"]);
        //$this->task("event:demo1");
        //$this->task("event:demo2");
        //$this->task("event:demo3");
    }
}

# 模擬一個發(fā)起的請求事件demo
private function pushDemo1Event($info)
{
    $sendInfoData = $this->getSendInfoFromInfo($info);

    //發(fā)起請求，這里做的真正的發(fā)起給對方的請求，你可以根據(jù)$sendInfo拼接各種事件或消息分類給對方
    $isSuccess = $this->sendInfo($sendInfoData);

    $data = array(
        "status"        => 1,
        "task_status"   => $isSuccess ? 1 : 2,
        "task_time"     => NOW_TIME
    );
    
    $db->save($data);

    //如果執(zhí)行失敗，重新插入一條記錄，并把at_time生成下次執(zhí)行的時間戳，這樣定時器根據(jù)at_time字段可以取出判斷執(zhí)行到當時的秒級
    if (!$isSuccess) {

        $tryCount = $db->where(...)->count();

        if ($tryCount >= 5) {
            return false;
        }

        $this->newTask($info, $tryCount, 30);
    }
}

private function newTask($info, $tryCount, $timeout = 30)
{
    $newTime = $this->nextTaskTime($info, $timeout);
    $data = array(
        ...
        "at_time" => newTime //執(zhí)行的時間（時間戳）
    );
    $db->add($data);
}

$timeType = $_GET["time_type"] ? : "now";
$lists = $this->getMsgAll($timeType);
if (!$lists) {
    return false;
}

foreach ($lists as $item) {
    if ($item["type"] == "1") {
        //根據(jù)業(yè)務生成我方真實訂單
        call_user_func_array(array($this, "add_real_order"), array($item));
    }
    if (in_array($item["type"], array(2,3))) {
        $this->push("push:toengineer", $item);  //指派或取消工程師
    }
    if ($item["type"] == "5") {
        $this->push("push:edit", $item);  //編輯
    }
    if ($item["type"] == "6") {
        $this->push("push:cancel", $item);  //取消
    }
    if ($item["type"] == "7" || $item["type"] == "8" || $item["type"] == "9") {
        $this->push("push:status", $item);  //簡單訂單狀態(tài)
    }
}

因為crontab是最小單位是分鐘，所以需要借助shell腳本來實現(xiàn)秒級執(zhí)行

#!/bin/bash
cd /web/project/src
for (( i = 1; i < 60; i = i + 1 ))
do
    #執(zhí)行php直接返回，不會阻塞 不要忘記最后面的 &
    $(/usr/bin/php index.php redis2mysql > /dev/null 2>&1 &)
    sleep 1
done

exit 0

#!/bin/bash
cd /web/project/src
for (( i = 1; i < 60; i = i + 1 ))
do
    $(/usr/bin/php index.php task_run > /dev/null 2>&1 &)
    sleep 1
done

exit 0

#!/bin/bash
cd /web/project/src
for (( i = 1; i < 60; i = i + 1 ))
do
    $(/usr/bin/php index.php task_run/tt/lt > /dev/null 2>&1 &)
    sleep 1
done

exit 0

* * * * * sh /web/project/src/shell/repair_build_order.sh
* * * * * sh /web/project/src/shell/repair_sync_second.sh
* * * * * sh /web/project/src/shell/repair_sync_lt_time.sh