站長資訊網(wǎng)
場景
在業(yè)務(wù)中有時(shí)會(huì)碰到延遲操作，如下單后半小時(shí)未支付則取消訂單、下單后十五分鐘未支付則發(fā)短信提醒等等。那這樣的需求如何去實(shí)現(xiàn)呢。
相關(guān)學(xué)習(xí)推薦：PHP編程從入門到精通
實(shí)現(xiàn)方式
- 第一個(gè)簡單的方式就是用一個(gè)后臺(tái)進(jìn)程死循環(huán)去查訂單，根據(jù)下單時(shí)間去做不同的操作
- 第二種就是使用消息隊(duì)列的定時(shí)消息，下單之后發(fā)送定時(shí)消息，不同的定時(shí)隊(duì)列去處理不同的邏輯
- 第三種可以使用框架提供的一些既有功能去做
實(shí)現(xiàn)代碼
我們以訂單創(chuàng)建15分鐘后未支付，給用戶發(fā)送郵件為場景進(jìn)行學(xué)習(xí)
準(zhǔn)備工作：
- 簡單的訂單表：order
- 各種需要的composer包
- rabbitMq本地服務(wù)
- 開通阿里云RocketMq服務(wù)
第一種
- 代碼邏輯很簡單就直接死循環(huán)就行了
- 啟動(dòng)這個(gè)腳本進(jìn)程，可以用supervisor配置
- 部分代碼
//創(chuàng)建訂單的邏輯/** * 隨機(jī)創(chuàng)建訂單 */$order = [ 'order_number' => mt_rand(100,10000).date("YmdHis"), 'user_id' => mt_rand(1, 100), 'order_amount' => mt_rand(100, 1000),]; /**@var $manager IlluminateDatabaseCapsuleManager **/ $conn = $manager;$insertResult = $conn::table("order") ->insert($order);print_r($insertResult);
延遲處理邏輯
while(true) { // 未支付訂單列表 $orderList = $conn::table("order") ->where("created_time", '<=', date("Y-m-d H:i:s", strtotime("-15 minutes"))) ->where('sended_need_pay_notify', '=', 2) ->where('status', '=', 1) ->select(['user_id', 'id']) ->orderBy("id", 'asc') ->get(); $orderList = json_decode(json_encode($orderList), true); foreach ($orderList as $orderInfo) { sendEmail($orderInfo['user_id']); $conn::table('order') ->where('id', '=', $orderInfo['id']) ->update(['sended_need_pay_notify' => 1]); logs("update-success-orderId-". $orderInfo['id']."-userId-".$orderInfo['user_id']); } sleep(10);}
執(zhí)行處理腳本
gaoz@nobodyMBP delay_mq_demo % php first_while_handler.php send email to 73 success ... 2020-06-24 11:37:36:update-success-orderId-3-userId-73
這種方式吧實(shí)現(xiàn)簡單，但是不優(yōu)雅，同時(shí)大批量訂單產(chǎn)生也會(huì)遇到問題。
第二種
- 比如使用阿里云的MQ服務(wù)，目前rocketMq與rabbitMq版本支持延遲消息，但是rabbit的延時(shí)消息收費(fèi)太高了
- 這里先使用rocketMq的延遲消息去實(shí)現(xiàn)
- 需要開通阿里云的服務(wù)
// 創(chuàng)建訂單的邏輯try { /** * 隨機(jī)創(chuàng)建訂單 */ $order = [ 'order_number' => mt_rand(100,10000).date("YmdHis"), 'user_id' => mt_rand(1, 100), 'order_amount' => mt_rand(100, 1000), ]; /**@var $manager IlluminateDatabaseCapsuleManager **/ $conn = $manager; $insertId = $conn::table("order") ->insertGetId($order); $body = json_encode(['order_id' => $insertId, 'created_time' => date("Y-m-d H:i:s")]); $publishMessage = new TopicMessage( $body ); // 設(shè)置消息KEY $publishMessage->setMessageKey("MessageKey"); // 定時(shí)消息, 定時(shí)時(shí)間為3分鐘后 $publishMessage->setStartDeliverTime(time() * 1000 + 3 * 60 * 1000); $result = $this->producer->publishMessage($publishMessage); print "Send mq message success. msgId is:" . $result->getMessageId() . ", bodyMD5 is:" . $result - >getMessageBodyMD5() . "n"; } catch (Exception $e) { print_r($e->getMessage() . "n"); }
消費(fèi)邏輯 同樣是在消費(fèi)者中處理
foreach ($messages as $message) { $receiptHandles[] = $message->getReceiptHandle(); $messageBody = $message->getMessageBody(); $orderInfo = json_decode($messageBody, true); if (!empty($orderInfo['order_id'])) { $orderId = $orderInfo['order_id']; /**@var $manager IlluminateDatabaseCapsuleManager * */ $conn = $manager; $orderInfo = $conn::table("order") ->select(['id', 'user_id']) ->where('id', '=', $orderId) ->where('status', '=', 1) ->first(); if (!empty($orderInfo)) { $orderInfo = json_decode(json_encode($orderInfo), true); sendEmail($orderInfo['user_id']); $conn::table('order') ->where('id', '=', $orderInfo['id']) ->update(['sended_need_pay_notify' => 1]); logs("update-success-orderId-" . $orderInfo['id'] . "-userId-" . $orderInfo['user_id']); } } }
啟動(dòng)生產(chǎn)一條消息
gaoz@nobodyMBP delay_mq_demo % php rocket_mq_handler_producer.php Send mq message success. msgId is:76CF2135696C3D4EAC698A9FA1E1879D, bodyMD5 is:63448B50AA7B8AF47B07AA7CE807E3D3 gaoz@nobodyMBP delay_mq_demo %
啟動(dòng)消費(fèi)者慢慢等待
gaoz@nobodyMBP delay_mq_demo % php rocket_mq_handler_consumer.php No message, contine long polling!RequestId:5EF752583441411C74869BA9 No message, contine long polling!RequestId:5EF7525B3441411C74869FE2 No message, contine long polling!RequestId:5EF7525E3441411C7486A42C No message, contine long polling!RequestId:5EF752613441411C7486A7D9 consume finish, messages:send email to 95 success ...2020-06-27 12:08:05:update-success-orderId-8-userId-95 Array( [0] => 76CF2135696C3D4EAC698A9FA1E1879D-MCAxNTkzMjY2NzkxNDM5IDMwMDAwMCAzIDAgYmpzaGFyZTUtMDggNSAw) ack
這種方式有現(xiàn)有的服務(wù)可以使用，減少開發(fā)時(shí)間
第三種 使用rabbitMq去實(shí)現(xiàn)
- 查閱文檔沒有找到rabbitMq支持延遲隊(duì)列的原生功能，但是可以通過消息的ttl+死信隊(duì)列實(shí)現(xiàn)
- 私信隊(duì)列就是用來存放沒有被消費(fèi)或者消費(fèi)失敗等消息的隊(duì)列
- 當(dāng)設(shè)置消息的有效期內(nèi)沒有被消費(fèi)消息就會(huì)被轉(zhuǎn)發(fā)到死信隊(duì)列
- 通過設(shè)置消息的有效期實(shí)現(xiàn)延時(shí)功能
// 生產(chǎn)者$exchange = 'order15min_notify_exchange'; $queue = 'order15minx_notify_queue';$dlxExchange = "dlx_order15min_exchange"; $dlxQueue = "dlx_order15min_queue"; $connection = new AMQPStreamConnection(getenv('RABBIT_HOST'), getenv('RABBIT_PORT'), getenv("RABBIT_USER"), getenv("RABBIT_PASS"), getenv("RABBIT_VHOST")); $channel = $connection->channel();$channel->exchange_declare($exchange, AMQPExchangeType::DIRECT, false, true, false); $channel->exchange_declare($dlxExchange, AMQPExchangeType::DIRECT, false, true, false);// 設(shè)置隊(duì)列的過期時(shí)間// 正常隊(duì)列$table = new PhpAmqpLibWireAMQPTable();// 消息有效期$table->set('x-message-ttl', 3*60*1000);$table->set("x-dead-letter-exchange", $dlxExchange);$channel->queue_declare($queue, false, true, false, false, false, $table);$channel->queue_bind($queue, $exchange);// 死信隊(duì)列$channel->queue_declare($dlxQueue, false, true, false, false, false);$channel->queue_bind($dlxQueue, $dlxExchange);/** * 隨機(jī)創(chuàng)建訂單 */$order = [ 'order_number' => mt_rand(100,10000).date("YmdHis"), 'user_id' => mt_rand(1, 100), 'order_amount' => mt_rand(100, 1000),];/**@var $manager IlluminateDatabaseCapsuleManager **/$conn = $manager;$insertId = $conn::table("order") ->insertGetId($order);$messageBody = json_encode(['order_id' => $insertId, 'created_time' => date("Y-m-d H:i:s")]); $message = new AMQPMessage($messageBody, array('content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT)); $channel->basic_publish($message, $exchange);
消費(fèi)者
$dlxExchange = "dlx_order15min_exchange";$dlxQueue = "dlx_order15min_queue"; $connection = new AMQPStreamConnection(getenv('RABBIT_HOST'), getenv('RABBIT_PORT'), getenv("RABBIT_USER"), getenv("RABBIT_PASS"), getenv("RABBIT_VHOST")); $channel = $connection->channel(); $channel->queue_declare($dlxQueue, false, true, false, false);$channel->exchange_declare($dlxExchange, AMQPExchangeType::DIRECT, false, true, false); $channel->queue_bind($dlxQueue, $dlxExchange);/** * @param PhpAmqpLibMessageAMQPMessage $message */function process_message($message){ echo "n--------n"; echo $message->body; echo "n--------n"; $orderInfo = json_decode($message->body, true); if (!empty($orderInfo['order_id'])) { $orderId = $orderInfo['order_id']; /**@var $conn IlluminateDatabaseCapsuleManager * */ $conn = getdb(); $orderInfo = $conn::table("order") ->select(['id', 'user_id']) ->where('id', '=', $orderId) ->where('status', '=', 1) ->first(); if (!empty($orderInfo)) { $orderInfo = json_decode(json_encode($orderInfo), true); sendEmail($orderInfo['user_id']); $conn::table('order') ->where('id', '=', $orderInfo['id']) ->update(['sended_need_pay_notify' => 1]); logs("update-success-orderId-" . $orderInfo['id'] . "-userId-" . $orderInfo['user_id']); } } $message->delivery_info['channel']->basic_ack( $message->delivery_info['delivery_tag']);}$channel->basic_consume($dlxQueue, $consumerTag, false, false, false, false, 'process_message');
啟動(dòng)消費(fèi)者
gaoz@nobodyMBP delay_mq_demo % php rabbit_mq_handler_consumer.php -------- {"order_id":7,"created_time":"2020-06-27 11:50:08"} -------- send email to 2 success ... 2020-06-27 11:56:55:update-success-orderId-7-userId-2
分別啟動(dòng)消費(fèi)者、生產(chǎn)者就可以了，這里面消息的流轉(zhuǎn)可以看到
消息先進(jìn)入到正常隊(duì)列，過期后進(jìn)入了死信隊(duì)列而被消費(fèi)
第四種
- 使用laravel自帶的Queue去實(shí)現(xiàn)
- 這里沒有整理詳細(xì)代碼，后面更新出來
- 可以查看官方文檔 隊(duì)列《Laravel 5.7 中文文檔》
代碼示例：github.com/nobody05/delay_mq_demo
