PHP は単純に遅延演算を実装します。
- coldplay.xixi転載
- 2020-07-03 17:41:245636ブラウズ
## シナリオ
#ビジネスでは、次のようなことがあります。注文後 30 分経過しても支払いが行われない場合は注文がキャンセルされる、注文後 15 分経過しても支払いが行われない場合にはテキスト メッセージのリマインダーが送信されるなど、遅延操作が発生する場合があります。では、そのような需要を実現するにはどうすればよいでしょうか? #関連する学習の推奨事項:PHP プログラミングの入門から熟練度まで#実装方法
1 つ目の簡単な方法は、バックグラウンド プロセスを使用して無限ループで注文を確認し、注文時間に応じてさまざまな操作を実行することです。
#2 つ目は、メッセージ キューのスケジュールされたメッセージを使用し、送信することです。注文後のスケジュールされたメッセージ。異なるタイミング キューは異なるロジックを処理します。- 3 番目の方法は、フレームワークによって提供されるいくつかの既存の関数を使用して実行できます。
- 実装コード
注文が作成されてから 15 分間支払いが行われなかった場合に、ユーザーに電子メールを送信するシナリオを使用します。 準備作業:
簡易オーダーフォーム: order
必要な各種コンポーザーパッケージrabbitMq ローカルサービス
Alibaba Cloud RocketMq サービスを開く- ##最初のコード
- ##コード ロジックは非常に単純で、無限ループに入るだけです
- このスクリプト プロセスを開始するには、スーパーバイザを使用して設定できます。
コードの一部//创建订单的逻辑/**
* 随机创建订单
*/$order = [
'order_number' => mt_rand(100,10000).date("YmdHis"),
'user_id' => mt_rand(1, 100),
'order_amount' => mt_rand(100, 1000),];
/**@var $manager Illuminate\Database\Capsule\Manager **/
$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);}
実行処理スクリプト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
たとえば、Alibaba Cloud の MQ サービスを使用すると、現在の rocketMq および RabbitMq バージョンは遅延メッセージをサポートしますが、 Rabbit の遅延メッセージの料金が高すぎます
ここでは、最初に rocketMq の遅延メッセージを使用して実装します
Alibaba Cloud サービスを有効にする必要があります// 创建订单的逻辑try
{
/**
* 随机创建订单
*/
$order = [
'order_number' => mt_rand(100,10000).date("YmdHis"),
'user_id' => mt_rand(1, 100),
'order_amount' => mt_rand(100, 1000),
];
/**@var $manager Illuminate\Database\Capsule\Manager **/
$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 );
// 设置消息KEY
$publishMessage->setMessageKey("MessageKey");
// 定时消息, 定时时间为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");
}消費ロジックも消費しています処理中 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 Illuminate\Database\Capsule\Manager * */
$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']);
}
}
}
メッセージの生成を開始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 %
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)
ack3 番目のタイプ
RabbitMq を使用して実装するドキュメントを調べた後、遅延キューをサポートする RabbitMq のネイティブ機能は見つかりませんでしたが、これは実現可能ですメッセージの ttl を通じてデッド レター キューの実装
プライベート メッセージ キューは、消費されなかったメッセージ、または消費に失敗したメッセージを保存するために使用されるキューです。
メッセージが期限内に消費されない場合設定したメッセージの有効期限が切れるとデッドレターキューに転送されます メッセージの有効期限を設定することで遅延機能を実現// 生产者$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);// 设置队列的过期时间// 正常队列$table = new \PhpAmqpLib\Wire\AMQPTable();// 消息有效期$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);// 死信队列$channel->queue_declare($dlxQueue, false, true, false, false, false);$channel->queue_bind($dlxQueue, $dlxExchange);/**
* 随机创建订单
*/$order = [
'order_number' => mt_rand(100,10000).date("YmdHis"),
'user_id' => mt_rand(1, 100),
'order_amount' => mt_rand(100, 1000),];/**@var $manager Illuminate\Database\Capsule\Manager **/$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);
Consumer$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 \PhpAmqpLib\Message\AMQPMessage $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 Illuminate\Database\Capsule\Manager * */
$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'); 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
##4番目の方法
##laravel独自のキューを使用します。実装するための
公式ドキュメントキュー「Laravel 5.7 Chinese Documentation」
##をご覧いただけます#コード例: github.com/nobody05/lay_mq_demo
以上がPHP は単純に遅延演算を実装します。の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。
声明：
この記事はlearnku.comで複製されています。侵害がある場合は、admin@php.cn までご連絡ください。