PHP は単純に遅延演算を実装します。-PHPチュートリアル-php.cn

PHP は単純に遅延演算を実装します。

## シナリオ

#ビジネスでは、次のようなことがあります。注文後 30 分経過しても支払いが行われない場合は注文がキャンセルされる、注文後 15 分経過しても支払いが行われない場合にはテキストメッセージのリマインダーが送信されるなど、遅延操作が発生する場合があります。では、そのような需要を実現するにはどうすればよいでしょうか?

#関連する学習の推奨事項:

PHP プログラミングの入門から熟練度まで

#実装方法

1 つ目の簡単な方法は、バックグラウンドプロセスを使用して無限ループで注文を確認し、注文時間に応じてさまざまな操作を実行することです。

#2 つ目は、メッセージキューのスケジュールされたメッセージを使用し、送信することです。注文後のスケジュールされたメッセージ。異なるタイミングキューは異なるロジックを処理します。

3 番目の方法は、フレームワークによって提供されるいくつかの既存の関数を使用して実行できます。
実装コード

注文が作成されてから 15 分間支払いが行われなかった場合に、ユーザーに電子メールを送信するシナリオを使用します。準備作業:

簡易オーダーフォーム: order

必要な各種コンポーザーパッケージ

rabbitMq ローカルサービス

Alibaba Cloud RocketMq サービスを開く

##最初のコード
##コードロジックは非常に単純で、無限ループに入るだけです
このスクリプトプロセスを開始するには、スーパーバイザを使用して設定できます。

コードの一部

//创建订单的逻辑/**
 * 随机创建订单
 */$order = [
    &#39;order_number&#39; => mt_rand(100,10000).date("YmdHis"),
    &#39;user_id&#39; => mt_rand(1, 100),
    &#39;order_amount&#39; => 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",  &#39;<=&#39;, date("Y-m-d H:i:s", strtotime("-15 minutes")))
        ->where(&#39;sended_need_pay_notify&#39;, &#39;=&#39;, 2)
        ->where(&#39;status&#39;, &#39;=&#39;, 1)
        ->select([&#39;user_id&#39;, &#39;id&#39;])
        ->orderBy("id", &#39;asc&#39;)
        ->get();
    $orderList = json_decode(json_encode($orderList), true);
    foreach ($orderList as $orderInfo) {
        sendEmail($orderInfo[&#39;user_id&#39;]);
        $conn::table(&#39;order&#39;)
            ->where(&#39;id&#39;, &#39;=&#39;, $orderInfo[&#39;id&#39;])
            ->update([&#39;sended_need_pay_notify&#39; => 1]);
        logs("update-success-orderId-". $orderInfo[&#39;id&#39;]."-userId-".$orderInfo[&#39;user_id&#39;]);
    }

    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 = [
                &#39;order_number&#39; => mt_rand(100,10000).date("YmdHis"),
                &#39;user_id&#39; => mt_rand(1, 100),
                &#39;order_amount&#39; => mt_rand(100, 1000),
            ];

            /**@var $manager Illuminate\Database\Capsule\Manager **/
            $conn = $manager;

            $insertId = $conn::table("order")
                ->insertGetId($order);

            $body = json_encode([&#39;order_id&#39; => $insertId, &#39;created_time&#39; => 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[&#39;order_id&#39;])) {
                    $orderId = $orderInfo[&#39;order_id&#39;];

                    /**@var $manager Illuminate\Database\Capsule\Manager * */
                    $conn = $manager;
                    $orderInfo = $conn::table("order")
                        ->select([&#39;id&#39;, &#39;user_id&#39;])
                        ->where(&#39;id&#39;, &#39;=&#39;, $orderId)
                        ->where(&#39;status&#39;, &#39;=&#39;, 1)
                        ->first();
                    if (!empty($orderInfo)) {
                        $orderInfo = json_decode(json_encode($orderInfo), true);
                        sendEmail($orderInfo[&#39;user_id&#39;]);
                        $conn::table(&#39;order&#39;)
                            ->where(&#39;id&#39;, &#39;=&#39;, $orderInfo[&#39;id&#39;])
                            ->update([&#39;sended_need_pay_notify&#39; => 1]);
                        logs("update-success-orderId-" . $orderInfo[&#39;id&#39;] . 
                        "-userId-" . $orderInfo[&#39;user_id&#39;]);
                    }
                }
            }

ログイン後にコピー

メッセージの生成を開始

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)
    ack

ログイン後にコピー

このメソッドは既存のサービスで使用でき、開発時間を短縮できます

3 番目のタイプ

RabbitMq を使用して実装する

ドキュメントを調べた後、遅延キューをサポートする RabbitMq のネイティブ機能は見つかりませんでしたが、これは実現可能ですメッセージの ttl を通じてデッドレターキューの実装

プライベートメッセージキューは、消費されなかったメッセージ、または消費に失敗したメッセージを保存するために使用されるキューです。

メッセージが期限内に消費されない場合設定したメッセージの有効期限が切れるとデッドレターキューに転送されますメッセージの有効期限を設定することで遅延機能を実現

// 生产者$exchange = &#39;order15min_notify_exchange&#39;;
$queue = &#39;order15minx_notify_queue&#39;;$dlxExchange = "dlx_order15min_exchange";
$dlxQueue = "dlx_order15min_queue";
$connection = new AMQPStreamConnection(getenv(&#39;RABBIT_HOST&#39;), getenv(&#39;RABBIT_PORT&#39;), 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(&#39;x-message-ttl&#39;, 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 = [
    &#39;order_number&#39; => mt_rand(100,10000).date("YmdHis"),
    &#39;user_id&#39; => mt_rand(1, 100),
    &#39;order_amount&#39; => mt_rand(100, 1000),];/**@var $manager Illuminate\Database\Capsule\Manager **/$conn = $manager;$insertId = $conn::table("order")
    ->insertGetId($order);$messageBody = json_encode([&#39;order_id&#39; => $insertId, &#39;created_time&#39; => date("Y-m-d H:i:s")]);
    $message = new AMQPMessage($messageBody, array(&#39;content_type&#39; => &#39;text/plain&#39;, &#39;delivery_mode&#39; => AMQPMessage::DELIVERY_MODE_PERSISTENT));
    $channel->basic_publish($message, $exchange);

ログイン後にコピー

Consumer

$dlxExchange = "dlx_order15min_exchange";$dlxQueue = "dlx_order15min_queue";
$connection = new AMQPStreamConnection(getenv(&#39;RABBIT_HOST&#39;), getenv(&#39;RABBIT_PORT&#39;), 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[&#39;order_id&#39;])) {
        $orderId = $orderInfo[&#39;order_id&#39;];

        /**@var $conn Illuminate\Database\Capsule\Manager * */
        $conn = getdb();
        $orderInfo = $conn::table("order")
            ->select([&#39;id&#39;, &#39;user_id&#39;])
            ->where(&#39;id&#39;, &#39;=&#39;, $orderId)
            ->where(&#39;status&#39;, &#39;=&#39;, 1)
            ->first();
        if (!empty($orderInfo)) {
            $orderInfo = json_decode(json_encode($orderInfo), true);
            sendEmail($orderInfo[&#39;user_id&#39;]);
            $conn::table(&#39;order&#39;)
                ->where(&#39;id&#39;, &#39;=&#39;, $orderInfo[&#39;id&#39;])
                ->update([&#39;sended_need_pay_notify&#39; => 1]);
            logs("update-success-orderId-" . $orderInfo[&#39;id&#39;] . "-userId-" . $orderInfo[&#39;user_id&#39;]);
        }

    }
    $message->delivery_info[&#39;channel&#39;]->basic_ack(
        $message->delivery_info[&#39;delivery_tag&#39;]);}$channel->basic_consume($dlxQueue, $consumerTag, false, false, false, false, &#39;process_message&#39;);

ログイン後にコピー

Start Consumer

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

ログイン後にコピー

Start Consumer と Production それだけです。ここでメッセージの流れを確認できます
##メッセージは最初に通常のキューに入り、期限切れ後にデッドレターに入ります。キューは消費されます。

##4番目の方法

##laravel独自のキューを使用します。実装するための
詳細なコードはここにはまとめていないので、後日更新します出てきます
公式ドキュメントキュー「Laravel 5.7 Chinese Documentation」

##をご覧いただけます#コード例: github.com/nobody05/lay_mq_demo
以上がPHP は単純に遅延演算を実装します。の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。