シンプルで信頼性の高いrabbitmqコンポーネントパッケージをphpで使用する方法

rabbitmq がプロジェクトで広く使用されている場合、ここに Rabbitmq の通常の機能の簡単な概要があり、コンポーザー パッケージにカプセル化されています。コンポーザー パッケージのアドレス、github アドレス、フォークは歓迎されますが、レベルが限られているため、必然的にバグがあります。 、貴重なご意見をお待ちしております。

easy-rabbitmq パッケージの紹介

php-amqplib/php-amqplib パッケージの二次カプセル化。使用できる共通関数のセットを提供します。ボックス生産ソリューションの概要。現在サポートされている機能リストは次のとおりです。

• 直接接続されたスイッチへのメッセージのプッシュ (遅延メッセージを含む)

• セクター スイッチへのメッセージのプッシュ (メッセージ)

• トピック スイッチへのメッセージのプッシュ (遅延メッセージを含む)

• 消費者による消費後の、サブスクリプション モードでの信頼性の高い消費失敗する 最大 5 回まで、支出を続けてみます。

• プル モードでの信頼性の高い消費。消費者は失敗後も消費の継続を試行します (最大 5 回まで)。

他のシナリオがある場合は、引き続き追加して後で繰り返してください。 ！

要件

インストール パッケージの PHP バージョン要件は、主に php-amqplib/php-amqplib パッケージ自体の要件に依存します。php5.0 のユーザーを考慮するため、では、php-amqplib/php - amqplib パッケージ V2.9.0 のバージョンを使用します。

具体的な要件については、ここを参照してください。

ただし、作者は php7.0 以降の使用を推奨しており、この開発キットもバージョン 7.0 で開発されています。

インストール

      composer require maweibinguo/easyrabbitmq

使用方法

ここでは、消費プロセスの信頼性を確保し、ワーカーの消費能力を高めるために、php スクリプト スーパーバイザーを組み合わせて使用​​することをお勧めします。スーパーバイザについて聞いたことがない場合は、ここをクリックして詳細を確認してください。

1. メッセージのプッシュ

1-1. 直接接続されたスイッチへのメッセージのプッシュ

      $config = [
          "host" => "127.0.0.1",
            "port" => "5672",
            "user" => "guest",
            "password" => "guest",
            "vhost" => "/",
            "channel_max_num" => 10,
      ];    
      $instance = RabbitMq::getInstance($config);
      
      //延迟消息,30 秒中后才会到达指定的交换机
      $instance->pushToDirect(
                        $msg = time(), //消息体内容
                        $exchange = "easy_direct_exchange", //交换机名称
                        $routingKey = "direct_test_queue", //消息的routingKey，consume(get) 方法到bingdingKey 要和routingKey保持一致
                        $delaySec = 30 //延迟秒数
      );

      //无延迟，推入到指定到直链交换机
      $instance->pushToDirect(
                        $msg = time(), //消息体内容
                        $exchange = "easy_direct_exchange", //交换机名称
                        $routingKey = "direct_test_queue", //消息的routingKey，consume(get) 方法到bingdingKey 要和routingKey保持一致
      );

1-2、メッセージをセクターにプッシュします。 switch

      $config = [
          "host" => "127.0.0.1",
            "port" => "5672",
            "user" => "guest",
            "password" => "guest",
            "vhost" => "/",
            "channel_max_num" => 10,
      ];    
      $instance = RabbitMq::getInstance($config);
      
      //延迟消息,30 秒中后才会到达指定的交换机
      $instance->pushToFanout(
                        $msg = time(), //消息体内容
                        $exchange = "easy_fanout_exchange", //交换机名称
                        $delaySec = 30 //延迟秒数
      );

      //无延迟，推入到指定到直链交换机
      $instance->pushToFanout(
                        $msg = time(), //消息体内容
                        $exchange = "easy_fanout_exchange" //交换机名称
      );

1-3、メッセージをトピックにプッシュします。 switch

      $config = [
          "host" => "127.0.0.1",
            "port" => "5672",
            "user" => "guest",
            "password" => "guest",
            "vhost" => "/",
            "channel_max_num" => 10,
      ];    
      $instance = RabbitMq::getInstance($config);
      
      //延迟消息,30 秒中后才会到达指定的交换机
      $instance->pushToTopic(
                        $msg = time(), //消息体内容
                        $exchange = "easy_topic_exchange", //交换机名称
                        /**
                         * routingKey 要同consum(get)方法的bindingKey相匹配
                         * bindingKey支持两种特殊的字符"*"、“#”，用作模糊匹配, 其中"*"用于匹配一个单词、“#”用于匹配多个单词(也可以是0个)
                         * 无论是bindingKey还是routingKey, 被"."分隔开的每一段独立的字符串就是一个单词, easy.topic.queue, 包含三个单词easy、topic、queue
                         */
                        $routingKey = "easy.topic.queue",
                        $delaySec = 30 //延迟秒数
      );

      //无延迟，推入到指定到直链交换机
      $instance->pushToTopic(
                        $msg = time(), //消息体内容
                        $exchange = "easy_topic_exchange", //交换机名称
                        $routingKey = "easy.topic.queue"    
      );

2、メッセージを消費します。

消費サポート自動再試行 (最大 5 回の再試行) では、消費が失敗するたびに、メッセージは消費キューに戻されます。失敗数が増加するにつれて、再試行時間は徐々に経過します。このクライアントがサポートする遅延戦略は次のとおりです:

失敗 1 回 (1 秒後に再度配信されます)、失敗 2 回 (配信されます) 2 秒後に再度配信されます）、後で再度配信されます）、3 回失敗（4 秒後に再度配信されます）、4 回失敗（8 秒後に再度配信されます）、5 回失敗（16 秒後に再度配信されます）

2-1. サブスクリプション モード

サブスクリプション モードでの信頼できる消費

      $config = [
          "host" => "127.0.0.1",
            "port" => "5672",
            "user" => "guest",
            "password" => "guest",
            "vhost" => "/",
            "channel_max_num" => 10,
      ];    
      $instance = RabbitMq::getInstance($config);
      $instance->consume(
            $queueName = "direct_test_queue",//订阅的队列名称
            $consumerTag = "c1",//消费标记
            $exchange = "easy_direct_exchange",//交换机名称
            $bindingKey = "direct_test_queue",//bindingkey，如果是直链交换机需要同routingKey保持一致
            $callback = function($msg){
                $body = $msg->body;
                file_put_contents("./test.log", "time => " . time() . "\t" . " body => " . $body . PHP_EOL , FILE_APPEND);
                //如果返回结果不绝对等于(===)true,那么将触发消息重试机制
                return false;
            },
            //5次消费消费失败后，失败消息将会投递到的队列名称
            $failedQueue = "easymq@failed"
      );

2-2. プル モード

プル モードでの信頼できる消費

      $config = [
          "host" => "127.0.0.1",
            "port" => "5672",
            "user" => "guest",
            "password" => "guest",
            "vhost" => "/",
            "channel_max_num" => 10,
      ];    
      $instance = RabbitMq::getInstance($config);
      $instance->get(
            $queue = "get_queue",
            $exchange = "easy_fanout_exchange",
            $bindingKey = "",
            $callback = function($msg){
                $body = $msg->body;
                file_put_contents("./test.log", "time => " . time() . "\t" . " body => " . $body . PHP_EOL , FILE_APPEND);
                //如果返回结果不绝对等于(===)true,那么将触发消息重试机制
                return false;
            },
            //5次消费消费失败后，失败消息将会投递到的队列名称
            $failedQueue = 'easymq@failed'
      );

推奨学習: php ビデオ チュートリアル

以上がシンプルで信頼性の高いrabbitmqコンポーネントパッケージをphpで使用する方法の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。