<ruby id="bdb3f"></ruby>

    <p id="bdb3f"><cite id="bdb3f"></cite></p>

      <p id="bdb3f"><cite id="bdb3f"><th id="bdb3f"></th></cite></p><p id="bdb3f"></p>
        <p id="bdb3f"><cite id="bdb3f"></cite></p>

          <pre id="bdb3f"></pre>
          <pre id="bdb3f"><del id="bdb3f"><thead id="bdb3f"></thead></del></pre>

          <ruby id="bdb3f"><mark id="bdb3f"></mark></ruby><ruby id="bdb3f"></ruby>
          <pre id="bdb3f"><pre id="bdb3f"><mark id="bdb3f"></mark></pre></pre><output id="bdb3f"></output><p id="bdb3f"></p><p id="bdb3f"></p>

          <pre id="bdb3f"><del id="bdb3f"><progress id="bdb3f"></progress></del></pre>

                <ruby id="bdb3f"></ruby>

                ??一站式輕松地調用各大LLM模型接口,支持GPT4、智譜、豆包、星火、月之暗面及文生圖、文生視頻 廣告
                >[info] topic 通配符模式 按照正則表達式模糊匹配:用消息的Routing Key與 Exchange和Queue 之間的Binding Key進行模糊匹配,如果匹配成功,將消息分發到該Queue。 Routing Key是一個句點號“. ”分隔的字符串(我們將被句點號“. ”分隔開的每一段獨立的字符串稱為一個單詞)。Binding Key與Routing Key一樣也是句點號“. ”分隔的字符串。Binding Key中可以存在兩種特殊字符“ \* ”與“#”,用于做模糊匹配,其中“\*”用于匹配一個單詞,“#”用于匹配多個單詞(可以是零個)。 ![](https://img.kancloud.cn/d5/8a/d58a1571e87e677f57dc584eb590a299_967x441.png) ![](https://img.kancloud.cn/37/ec/37eccf9f1c43019dfd821f8d8129fde0_1002x396.png) 1. `rabbitmq.php` 配置文件信息 ~~~ <?php // rabbitmq 配置信息 return [ # 連接信息 'amqp' => [ 'host' => '127.0.0.1', 'port'=>'5672', 'user'=>'guest', 'password'=>'guest', 'vhost'=>'/' ], # 通配符隊列 'topic_queue' => [ 'exchange_name' => 'topic_exchange', 'exchange_type'=>'topic',# 通配符模式 'queue_name' => 'topic_queue', 'route_key' => 'topic.goods.order',# 模糊匹配 'consumer_tag' => 'topic' ], # 商品隊列 'goods_queue' => [ 'exchange_name' => 'goods_topic_exchange', 'exchange_type'=>'topic',# 通配符模式 'queue_name' => 'goods_queue', 'route_key' => '*.goods.*',# 模糊匹配 'consumer_tag' => 'goods' ], # 訂單隊列 'order_queue' => [ 'exchange_name' => 'order_topic_exchange', 'exchange_type'=>'topic',# 通配符模式 'queue_name' => 'order_queue', 'route_key' => '#.order',# 模糊匹配 'consumer_tag' => 'order' ], ]; ~~~ 2. `Consumer` 消費者 ~~~ <?php declare (strict_types = 1); namespace app\common\service\rabbitmq; //use app\common\model\test\MessageModel; use PhpAmqpLib\Connection\AMQPStreamConnection; use think\facade\Log; /** * 消費者 */ class Consumer { /** * 商品消費者 * @return \think\Response */ public function goods() { $mqConfig = config('rabbitmq.amqp'); $goodsConfig = config('rabbitmq.goods_queue'); // 創建連接 $connection = new AMQPStreamConnection( $mqConfig['host'], $mqConfig['port'], $mqConfig['user'], $mqConfig['password'], $mqConfig['vhost'] ); // 連接信道 $channel = $connection->channel(); // 設置消費者(Consumer)客戶端同時只處理一條隊列 // 這樣是告訴RabbitMQ,再同一時刻,不要發送超過1條消息給一個消費者(Consumer), // 直到它已經處理了上一條消息并且作出了響應。這樣,RabbitMQ就會把消息分發給下一個空閑的消費者(Consumer)。 // 消費者端要把自動確認autoAck設置為false,basic_qos才有效。 // 流量控制 $channel->basic_qos(0, 1, false); // 同樣是創建路由和隊列,以及綁定路由隊列,注意要跟producer(生產者)的一致 // 這里其實可以不用設置,但是為了防止隊列沒有被創建所以做的容錯處理 // 創建交換機(Exchange) // $channel->exchange_declare($goodsConfig['exchange_name'], $goodsConfig['exchange_type'], false, true, false); // 創建隊列 $channel->queue_declare($goodsConfig['queue_name'], false, true, false, false); // 綁定隊列和交換機 // $channel->queue_bind($goodsConfig['queue_name'], $goodsConfig['exchange_name'], $goodsConfig['route_key']); /** * 消費消息 * * queue: queue_name 被消費的隊列名稱 * consumer_tag: consumer_tag 消費者客戶端身份標識,用于區分多個客戶端 * no_local: false 這個功能屬于AMQP的標準,但是RabbitMQ并沒有做實現 * no_ack: true 收到消息后,是否不需要回復確認即被認為被消費 * exclusive: false 是否排他,即這個隊列只能由一個消費者消費。適用于任務不允許進行并發處理的情況下 * nowait: false 不返回執行結果,但是如果排他開啟的話,則必須需要等待結果的,如果兩個一起開就會報錯 * callback: $callback 回調邏輯處理函數 */ $channel->basic_consume($goodsConfig['queue_name'], $goodsConfig['consumer_tag'], false, false, false, false, array($this, 'process_message')); // 退出,執行shutdown來關閉通道與連接 register_shutdown_function(array($this, 'shutdown'), $channel, $connection); // 阻塞隊列監聽事件 while (count($channel->callbacks)) { $channel->wait(); } } /** * 訂單消費者 * @return \think\Response */ public function order() { $mqConfig = config('rabbitmq.amqp'); $orderConfig = config('rabbitmq.order_queue'); // 創建連接 $connection = new AMQPStreamConnection( $mqConfig['host'], $mqConfig['port'], $mqConfig['user'], $mqConfig['password'], $mqConfig['vhost'] ); // 連接信道 $channel = $connection->channel(); // 設置消費者(Consumer)客戶端同時只處理一條隊列 // 這樣是告訴RabbitMQ,再同一時刻,不要發送超過1條消息給一個消費者(Consumer), // 直到它已經處理了上一條消息并且作出了響應。這樣,RabbitMQ就會把消息分發給下一個空閑的消費者(Consumer)。 // 消費者端要把自動確認autoAck設置為false,basic_qos才有效。 // 流量控制 $channel->basic_qos(0, 1, false); // 同樣是創建路由和隊列,以及綁定路由隊列,注意要跟producer(生產者)的一致 // 這里其實可以不用設置,但是為了防止隊列沒有被創建所以做的容錯處理 // 創建交換機(Exchange) // $channel->exchange_declare($orderConfig['exchange_name'], $orderConfig['exchange_type'], false, true, false); // 創建隊列 $channel->queue_declare($orderConfig['queue_name'], false, true, false, false); // 綁定隊列和交換機 // $channel->queue_bind($orderConfig['queue_name'], $orderConfig['exchange_name'], $orderConfig['route_key']); /** * 消費消息 * * queue: queue_name 被消費的隊列名稱 * consumer_tag: consumer_tag 消費者客戶端身份標識,用于區分多個客戶端 * no_local: false 這個功能屬于AMQP的標準,但是RabbitMQ并沒有做實現 * no_ack: true 收到消息后,是否不需要回復確認即被認為被消費 * exclusive: false 是否排他,即這個隊列只能由一個消費者消費。適用于任務不允許進行并發處理的情況下 * nowait: false 不返回執行結果,但是如果排他開啟的話,則必須需要等待結果的,如果兩個一起開就會報錯 * callback: $callback 回調邏輯處理函數 */ $channel->basic_consume($orderConfig['queue_name'], $orderConfig['consumer_tag'], false, false, false, false, array($this, 'process_message')); // 退出,執行shutdown來關閉通道與連接 register_shutdown_function(array($this, 'shutdown'), $channel, $connection); // 阻塞隊列監聽事件 while (count($channel->callbacks)) { $channel->wait(); } } /** * 消息處理 * @param $message */ public function process_message($message) { if ($message->body !== 'quit') { $messageBody = json_decode($message->body); // 自定義的消息類型 if (!isset($messageBody->message_type)) { Log::write("error data:" . $message->body, 2); } else { // $messageModel = new MessageModel(); try { // 消息 Log::write("message_data:" . json_encode($message, JSON_UNESCAPED_UNICODE)); $body = json_decode($message->body, true); dump("消費消息如下:"); dump($body); // $messageModel->test($body); } catch (\Think\Exception $e) { Log::write($e->getMessage(), 2); Log::write(json_encode($message), 2); } catch (\PDOException $pe) { Log::write($pe->getMessage(), 2); Log::write(json_encode($message), 2); } } } // 手動確認ack,確保消息已經處理 $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']); // Send a message with the string "quit" to cancel the consumer. if ($message->body === 'quit') { $message->delivery_info['channel']->basic_cancel($message->delivery_info['consumer_tag']); } } /** * 關閉進程 * @param $channel * @param $connection */ public function shutdown($channel, $connection) { // 關閉信道 $channel->close(); // 關閉連接 $connection->close(); } } ~~~ 3. `Producer` 生產者 ~~~ <?php declare (strict_types = 1); namespace app\common\service\rabbitmq; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; /** * rabbitmq 生產者 */ class Producer { // 連接 protected $connection; // 管道 protected $channel; // 配置內容(基本) protected $mqConfig; // 配置內容(通配符) protected $topicConfig; protected $goodsConfig; protected $orderConfig; public function __construct() { $this->mqConfig = config('rabbitmq.amqp'); $this->topicConfig = config('rabbitmq.topic_queue'); $this->goodsConfig = config('rabbitmq.goods_queue'); $this->orderConfig = config('rabbitmq.order_queue'); // 創建連接 $this->connection = new AMQPStreamConnection( $this->mqConfig['host'], $this->mqConfig['port'], $this->mqConfig['user'], $this->mqConfig['password'], $this->mqConfig['vhost'] // 虛擬主機(起到消息隔離的作用) ); // 創建通道 $this->channel = $this->connection->channel(); } /** * 發送消息 * @param $data 消息內容 */ public function send($data) { /* * 流量控制 Specifies QoS * 消費者在開啟acknowledge的情況下,對接收到的消息需要異步對消息進行確認 * 由于消費者自身處理能力有限,從rabbitmq獲取一定數量的消息后,希望rabbitmq不再將隊列中的消息推送過來, * 當對消息處理完后(即對消息進行了ack,并且有能力處理更多的消息)再接收來自隊列的消息 * @param int $prefetch_size 最大unacked消息的字節數 * @param int $prefetch_count 最大unacked消息的條數 * @param bool $a_global 上述限制的限定對象,false限制單個消費者,true限制整個通道 * @return mixed */ $this->channel->basic_qos(0, 1, false); /** * creatingAQueue * name: hello 隊列名稱 * passive: false 如果設置true存在則返回OK,否則就報錯。設置false存在返回OK,不存在則自動創建 * durable: true 是否持久化,設置false是存放到內存中的,RabbitMQ重啟后會丟失;設置true,則代表是一個持久化的隊列,服務重啟后也會存在,因為服務會把持久化的queue存放到磁盤上當服務重啟的時候,會重新加載之前被持久化的queue * exclusive: false 是否排他,指定該選項為true則隊列只對當前連接有效,連接斷開后自動刪除 * auto_delete: false 是否自動刪除,當最后一個消費者斷開連接之后隊列是否自動被刪除 */ $this->channel->queue_declare($this->topicConfig['queue_name'], false, true, false, false); /** * 創建交換機(Exchange) * name: hello 交換機名稱 * type: direct 交換機類型,分別為direct/topic/topic,參考另外文章的Exchange Type說明。 * passive: false Return OK if set true exists, otherwise an error is reported. If false exists, OK is returned. If it does not exist, it is created automatically * durable: false 是否持久化,設置false是存放到內存中的,RabbitMQ重啟后會丟失 * auto_delete: false 是否自動刪除,當最后一個消費者斷開連接之后隊列是否自動被刪除 */ $this->channel->exchange_declare($this->topicConfig['exchange_name'], $this->topicConfig['exchange_type'], false, true, false); // 綁定消息交換機和隊列 $this->channel->queue_bind($this->topicConfig['queue_name'], $this->topicConfig['exchange_name'], $this->topicConfig['route_key']); $this->channel->queue_bind($this->goodsConfig['queue_name'], $this->topicConfig['exchange_name'], $this->goodsConfig['route_key']); $this->channel->queue_bind($this->orderConfig['queue_name'], $this->topicConfig['exchange_name'], $this->orderConfig['route_key']); // 將要發送數據變為json字符串 $messageBody = json_encode($data, JSON_UNESCAPED_UNICODE); /** * 創建AMQP消息類型 * delivery_mode 消息是否持久化 * AMQPMessage::DELIVERY_MODE_NON_PERSISTENT 不持久化 * AMQPMessage::DELIVERY_MODE_PERSISTENT 持久化 */ $message = new AMQPMessage($messageBody, [ 'content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT ]); /** * 發送消息 * msg: $message AMQP消息內容 * exchange: vckai_exchange 交換機名稱 * routing_key: hello 路由key */ $this->channel->basic_publish($message, $this->topicConfig['exchange_name'], $this->topicConfig['route_key']); // 關閉連接 $this->stop(); } // 關閉連接 public function stop() { $this->channel->close(); $this->connection->close(); } } ~~~ 4. 創建 `goods` 和 `order` 自定義命令,作為消費者。 同 fanout 訂閱/廣播模式 5. 配置 `config/console.php` 命令行 同 fanout 訂閱/廣播模式 6. 路由發送消息 `route/app.php` 同 fanout 訂閱/廣播模式 7. 啟動消費者 同 fanout 訂閱/廣播模式 8. 生產者發送消息 同 fanout 訂閱/廣播模式 ![](https://img.kancloud.cn/24/01/24015d8b7d220ef3ed9cdc9e500a02a1_694x287.png) ![](https://img.kancloud.cn/8e/86/8e86ce4f1d6efe5ee09a25e9eeed4f4d_707x283.png)
                  <ruby id="bdb3f"></ruby>

                  <p id="bdb3f"><cite id="bdb3f"></cite></p>

                    <p id="bdb3f"><cite id="bdb3f"><th id="bdb3f"></th></cite></p><p id="bdb3f"></p>
                      <p id="bdb3f"><cite id="bdb3f"></cite></p>

                        <pre id="bdb3f"></pre>
                        <pre id="bdb3f"><del id="bdb3f"><thead id="bdb3f"></thead></del></pre>

                        <ruby id="bdb3f"><mark id="bdb3f"></mark></ruby><ruby id="bdb3f"></ruby>
                        <pre id="bdb3f"><pre id="bdb3f"><mark id="bdb3f"></mark></pre></pre><output id="bdb3f"></output><p id="bdb3f"></p><p id="bdb3f"></p>

                        <pre id="bdb3f"><del id="bdb3f"><progress id="bdb3f"></progress></del></pre>

                              <ruby id="bdb3f"></ruby>

                              哎呀哎呀视频在线观看