PHP消息隊(duì)列實(shí)現(xiàn)及應(yīng)用詳解【隊(duì)列處理訂單系統(tǒng)和配送系統(tǒng)】

本文實(shí)例講述了PHP消息隊(duì)列實(shí)現(xiàn)及應(yīng)用。分享給大家供大家參考，具體如下：

在互聯(lián)網(wǎng)項(xiàng)目開(kāi)發(fā)者經(jīng)常會(huì)遇到『給用戶群發(fā)短信』、『訂單系統(tǒng)有大量的日志需要記錄』或者在秒殺業(yè)務(wù)的時(shí)候服務(wù)器無(wú)法承受瞬間并發(fā)的壓力。

這種情況下，我們?cè)趺幢ＷC系統(tǒng)正常有效的運(yùn)行呢？

這個(gè)時(shí)候，我們可以引入一個(gè)叫『消息隊(duì)列』的概念來(lái)解決上面的需求。

消息隊(duì)列的概念、原理和場(chǎng)景

在高并發(fā)的時(shí)候，程序往往無(wú)法做到及時(shí)的處理。我們引入一個(gè)中間的系統(tǒng)，來(lái)進(jìn)行分流和減壓。

所以從本質(zhì)上講：消息隊(duì)列就是一個(gè)隊(duì)列結(jié)構(gòu)的中間件。也就是說(shuō)，你把消息和內(nèi)容放入這個(gè)容器之后就可以直接返回，不用等它后期處理的結(jié)果。另外會(huì)有一個(gè)程序，讀取這些數(shù)據(jù)并按照順序處理。

1、隊(duì)列結(jié)構(gòu)的中間件
2、消息放入后，不必立即處理
3、由訂閱者/消費(fèi)者按順序處理

也就是說(shuō)：當(dāng)遇到一個(gè)比較大或者耗時(shí)比較長(zhǎng)的環(huán)節(jié)的時(shí)候，而同時(shí)你的業(yè)務(wù)又不需要立即知道這個(gè)環(huán)節(jié)的結(jié)果，使用消息隊(duì)列是好的選擇。

核心結(jié)構(gòu)如下面：

消息隊(duì)列 適用場(chǎng)景

一、數(shù)據(jù)需要冗余的時(shí)候
比如訂單系統(tǒng)中，后續(xù)需要進(jìn)行數(shù)據(jù)的轉(zhuǎn)換和記錄。消息隊(duì)列可以把這些數(shù)據(jù)持久化的存儲(chǔ)在隊(duì)列中，然后由訂單后期處理程序進(jìn)行處理，處理完成之后再把這條記錄從隊(duì)列中刪除。

二、系統(tǒng)的解耦
消息隊(duì)列解決了2套系統(tǒng)之間深度耦合的問(wèn)題。
使用消息隊(duì)列后，入隊(duì)的系統(tǒng)和出隊(duì)的系統(tǒng)沒(méi)有直接的關(guān)系。
入隊(duì)系統(tǒng)和出隊(duì)系統(tǒng)，其中一個(gè)崩潰之后不會(huì)影響另外一個(gè)的正常運(yùn)行。

三、流量削峰
就是秒殺和搶購(gòu)的時(shí)候，會(huì)出現(xiàn)明顯的流量劇增，對(duì)服務(wù)器的壓力非常大。
實(shí)際項(xiàng)目開(kāi)發(fā)中，配合緩存來(lái)使用消息隊(duì)列，一種很好的方案。

四、異步通信
消息隊(duì)列本身就實(shí)現(xiàn)了程序的異步操作，因此只要適合于異步的場(chǎng)景都可以使用消息隊(duì)列

五、擴(kuò)展性
比如訂單系統(tǒng)，訂單入隊(duì)之后，后期或許還有財(cái)務(wù)系統(tǒng)處理，但是如果還要加一個(gè)配貨系統(tǒng)。
只需要讓這個(gè)配貨系統(tǒng) 訂閱這個(gè) 消息隊(duì)列 即可。

六、排序保證
在有些場(chǎng)景下，數(shù)據(jù)的處理順序是非常重要的，隊(duì)列本身就可以做成單線程的單進(jìn)單出的系統(tǒng)。
從而有效的保證數(shù)據(jù)按照順序進(jìn)行處理。

常見(jiàn) 隊(duì)列實(shí)現(xiàn) 的優(yōu)缺點(diǎn)

隊(duì)列介質(zhì)：

Mysql：可靠性高、易實(shí)現(xiàn)、速度慢
Redis：速度快，單條大消息包時(shí)效率低
消息系統(tǒng)：專業(yè)性強(qiáng)、可靠，學(xué)習(xí)成本高(比如：RabbtiMQ)

消息處理的觸發(fā)機(jī)制：

死循環(huán)方式讀?。阂讓?shí)現(xiàn)，故障時(shí)無(wú)法及時(shí)恢復(fù)；
定時(shí)任務(wù)：壓力均分，有處理量上限。(最大的缺陷：定位任務(wù)時(shí)間的間隔和處理的數(shù)據(jù)需要精準(zhǔn)把握，不能上一個(gè)任務(wù)還沒(méi)有處理完成，下一個(gè)認(rèn)為就已經(jīng)啟動(dòng)了)
守護(hù)進(jìn)程：類似于PHP-FPM和PHP-CGI，需要shell知識(shí)

解耦案列：隊(duì)列處理 訂單系統(tǒng)和配送系統(tǒng)

我們?cè)谇懊媪私膺^(guò)消息隊(duì)列的使用場(chǎng)景

這里，我們要來(lái)處理其中一個(gè)場(chǎng)景：系統(tǒng)的解耦。

在電商項(xiàng)目中，當(dāng)客戶提交了一個(gè)訂單之后，客戶在個(gè)人中心可以看到訂單處于配送中。
這個(gè)時(shí)候就要參與進(jìn)來(lái)一個(gè)系統(tǒng)，叫做『配送系統(tǒng)』。如果我們?cè)谧黾軜?gòu)的時(shí)候，把訂單系統(tǒng)和配送系統(tǒng)設(shè)計(jì)在一起的話就會(huì)出現(xiàn)一些問(wèn)題：訂單系統(tǒng)的壓力比較大，但是配送系統(tǒng)沒(méi)有必要對(duì)這些壓力做及時(shí)的反應(yīng)；我們不需要訂單系統(tǒng)出現(xiàn)故障之后導(dǎo)致配送系統(tǒng)故障。

所以我們需要把這2個(gè)系統(tǒng)分開(kāi)，通過(guò)一個(gè)中間的隊(duì)列表來(lái)實(shí)現(xiàn)這2個(gè)系統(tǒng)的溝通。

如下圖架構(gòu)：

具體到我們的程序代碼大致邏輯如下圖：

大致流程：order.php來(lái)接收用戶訂單，生成訂單號(hào)并對(duì)訂單進(jìn)行處理（訂單系統(tǒng)）；在訂單系統(tǒng)會(huì)把配送系統(tǒng)所需要的數(shù)據(jù)放入隊(duì)列表中；我們的配送系統(tǒng)goods.php會(huì)有個(gè)定時(shí)腳本每分鐘執(zhí)行一次，處理隊(duì)列表中的數(shù)據(jù)。

簡(jiǎn)單設(shè)計(jì)隊(duì)列表order_queue：

CREATE TABLE `order_queue` (
 `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
 `order_id` int(11) unsigned NOT NULL COMMENT '訂單ID(從訂單系統(tǒng)來(lái)的)',
 `user_info` varchar(255) NOT NULL DEFAULT '' COMMENT '可以是用戶手機(jī)號(hào)/用戶id等(這里只是演示)',
 `created_at` datetime NOT NULL COMMENT '訂單創(chuàng)建時(shí)間',
 `updated_at` datetime NOT NULL COMMENT '本記錄最后處理完成時(shí)間',
 `status` tinyint(2) NOT NULL COMMENT '0未處理，1已處理，2處理中',
 PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

mysql訂單隊(duì)列

前面我們已經(jīng)分析清楚了邏輯，剩下的就是代碼實(shí)現(xiàn)了。

注意：我這里只是演示代碼，單純?yōu)榱苏故緦?shí)現(xiàn)過(guò)程。

1、接收訂單，處理訂單order.php

?php
// 這個(gè)文件是用來(lái)接收用戶的訂單信息 并寫入隊(duì)列的一個(gè)文件
if(!empty($_GET['user_info'])){
  // 驗(yàn)證 過(guò)濾 接收的數(shù)據(jù)
  // todo...
  // 這里是應(yīng)該首先是訂單中心的處理流程
  // 因?yàn)橛唵蜗到y(tǒng)是一套單獨(dú)的系統(tǒng) 這里就不編寫這個(gè)系統(tǒng)了
  // todo...
  $order_id = rand(100000,99999); // 正常的訂單號(hào)從 訂單系統(tǒng)來(lái)，我們這里只是演示
  // 把配送系統(tǒng)需要的訂單數(shù)據(jù)存入隊(duì)列表中
  $insert_data = array(
    'order_id'=>$order_id,
    'user_info'=>$_GET['user_info'],
    'created_at'=>date('Y-m-d H:i:s',time()),
    'status'=>0
  );
  // 把上面的數(shù)據(jù) 插入到order_queue表中
  // insert into order_queue
}

2、配送系統(tǒng)goods.php

?php
// 這個(gè)文件主要是配送系統(tǒng)處理隊(duì)列表中的訂單并進(jìn)行標(biāo)記的文件
//分析：
//第一步：先把要處理的記錄更新為『等待處理』
//第二步：選擇剛剛標(biāo)記為『等待處理』的記錄，然后進(jìn)行配送系統(tǒng)的處理
//第三步：把上面前面處理過(guò)的程序標(biāo)記『已完成』
/////////////////////這里很重要，你一定要明白哦//////////////////////////////////////////////
//疑問(wèn)：為什么不直接處理最后更新為『已完成』，多了先標(biāo)記為『等待處理』？
//這是因?yàn)榕渌拖到y(tǒng)很可能不是及時(shí)完成的，它中間會(huì)有一段處理的時(shí)間，如果還在處理中有其他程序來(lái)進(jìn)行讀取和操作，就沖突了。
//這樣設(shè)計(jì)其實(shí)也是一個(gè)鎖的機(jī)制
//1、
$waiting = array('status'=>0);
$lock = array('status'=>2);
//把狀態(tài)為0的記錄標(biāo)記為2，每次更新3條（具體每次幾條看情況）
$sql = "update order_queue set status=2 where status=0 limit 3";
//2、
if(上面update成功){
  // 選擇出要處理訂單內(nèi)容
  // select * from order_queue where status = 2;
  // 然后由配貨系統(tǒng)進(jìn)行處理
  // todo...
  //3、處理完成把訂單狀態(tài)更新為已完成
  $success = array(
    'status'=>1,
    'updated_at'=>date('Y-m-d H:i:s',time())
  );
}else{
  echo 'All Finished';
}

3、linux服務(wù)器 定時(shí)任務(wù)

寫個(gè)shell腳本：goods.sh

#!/bin/bash
date "+%G-%m-%d %H:%M:%S"
cd /var/www/
php goods.php

這個(gè)腳本就是去執(zhí)行orders.php這個(gè)程序的。

在linux服務(wù)器部署定時(shí)任務(wù)：

crontab -e

*/1 * * * * /var/www/goods.sh >> /var/www/goods_shell.log 2>$1

每分鐘執(zhí)行一次goods.sh文件，并記錄日志到goods_shell.log文件(在對(duì)應(yīng)目錄新建該文件)

更多關(guān)于PHP相關(guān)內(nèi)容感興趣的讀者可查看本站專題：《PHP數(shù)據(jù)結(jié)構(gòu)與算法教程》、《php程序設(shè)計(jì)算法總結(jié)》、《php字符串(string)用法總結(jié)》、《PHP數(shù)組(Array)操作技巧大全》、《PHP常用遍歷算法與技巧總結(jié)》及《PHP數(shù)學(xué)運(yùn)算技巧總結(jié)》

希望本文所述對(duì)大家PHP程序設(shè)計(jì)有所幫助。