PHP implementiert einfach eine Verzögerungsoperation

                                                       

Szenario

Manchmal im Geschäft Du Es kommt zu Verzögerungen bei Vorgängen, z. B. der Stornierung der Bestellung, wenn die Zahlung nicht eine halbe Stunde nach der Bestellung erfolgt, dem Versenden einer SMS-Erinnerung, wenn die Zahlung fünfzehn Minuten nach der Bestellung nicht erfolgt usw. Wie lässt sich eine solche Forderung realisieren?

Verwandte Lernempfehlungen: PHP-Programmierung vom Einstieg bis zur Beherrschung

Implementierungsmethoden

• Die erste einfache Möglichkeit besteht darin, einen Hintergrundprozess zu verwenden, um die Bestellung in einer Endlosschleife zu überprüfen und je nach Bestellzeit verschiedene Vorgänge auszuführen
• Die zweite Möglichkeit besteht darin, die geplante Nachricht der Nachrichtenwarteschlange zum Senden zu verwenden Geplante Nachricht nach der Bestellung, verschiedene Timing-Warteschlangen behandeln unterschiedliche Logik
• Die dritte Methode kann mithilfe einiger vorhandener Funktionen durchgeführt werden, die vom Framework bereitgestellt werden

Implementierungscode

Wir haben das Szenario verwendet, eine E-Mail an den Benutzer zu senden, wenn die Bestellung 15 Minuten nach der Erstellung der Bestellung nicht bezahlt wurde 🎜>Vorbereitung:

Einfaches Bestellformular: BestellungVerschiedene erforderliche Composer-Pakete

RabbitMq-Lokaldienst
1. Offener Alibaba Cloud RocketMq-Dienst
4. Der erste Typ

Die Codelogik ist sehr einfach, einfach eine Endlosschleife durchlaufen Starten Sie diesen Skriptprozess und konfigurieren Sie ihn mit dem Supervisor

Teil des Codes

• //创建订单的逻辑/**
   * 随机创建订单
   */$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);

• Verzögerte Verarbeitungslogik

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);}

• Ausführung des Verarbeitungsskripts

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

Diese Methode ist einfach zu implementieren, aber nicht elegant, und es können große Mengen gleichzeitig bestellt werden. Es werden auch Probleme auftreten.

Der zweite Typ

Zum Beispiel unterstützen die aktuellen Versionen von RocketMq und RabbitMq bei Verwendung des MQ-Dienstes von Alibaba Cloud verzögerte Nachrichten. aber die Gebühren für verzögerte Nachrichten von Rabbit sind zu hochHier verwenden wir zunächst die verzögerte Nachricht von RocketMq zur Implementierung

Sie müssen Alibaba Cloud-Dienste aktivieren

• // 创建订单的逻辑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");
          }

• Die Verbrauchslogik ist verbraucht auch Prozessor

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']);
                    }
                }
            }

• Beginne mit der Erstellung einer Nachricht

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 %

Starte den Verbraucher und warte langsam

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

Diese Methode kann vorhandene Dienste nutzen und die Entwicklungszeit verkürzen

Der dritte Typ

Verwenden Sie RabbitMq zur Implementierung

Bei der Überprüfung der Dokumentation wurde die native Funktion von RabbitMq zur Unterstützung der Verzögerungswarteschlange nicht gefunden, sie kann jedoch über ttl+ erreicht werden der Implementierung der Warteschlange für unzustellbare NachrichtenDie private Nachrichtenwarteschlange ist eine Warteschlange zum Speichern von Nachrichten, die nicht verbraucht wurden oder nicht verbraucht werden konnten

Wenn die Nachricht nicht innerhalb des Gültigkeitszeitraums verbraucht wird der eingestellten Nachricht wird sie an die Warteschlange für unzustellbare Nachrichten weitergeleitet
• Realisieren Sie die Verzögerungsfunktion, indem Sie die Gültigkeitsdauer der Nachricht festlegen

• // 生产者$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');

• Starten Sie den Verbraucher

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

Starten Sie den Verbraucher bzw. die Produktion. Den Nachrichtenfluss können Sie hier sehen

Die Nachricht Betritt zuerst die normale Warteschlange und tritt dann nach Ablauf in die Warteschlange für tote Buchstaben ein und wird verbraucht

Die vierte Methode

Verwenden Sie zur Implementierung die eigene Warteschlange von Laravel Der detaillierte Code ist hier nicht organisiert und wird später aktualisiert. Kommen Sie vorbei

, um die offizielle Dokumentenwarteschlange „Laravel 5.7 Chinese Documentation“ anzuzeigen
• Codebeispiel: Github .com/nobody05/delay_mq_demo

Das obige ist der detaillierte Inhalt vonPHP implementiert einfach eine Verzögerungsoperation. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!