1.連接

// 連接服務，獲取句柄 $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');$connection->isConnected() or die("Cannot connect to the broker!\n");$channel = $connection->channel();

?2.定義

//direct：在direct類型的exchange中，只有這兩個routingkey完全相同，exchange才會選擇對應的binging進行消息路由。 //fanout：此exchange的路由規則很簡單直接將消息路由到所有綁定的隊列中，無須對消息的routingkey進行匹配操作。 //topic：此類型exchange和上面的direct類型差不多，這里的routingkey可以有通配符：'*','#' ，其中'*'表示匹配一個單詞， // '#'則表示匹配沒有或者多個單詞，對 key 進行模式匹配，比如 ab* 可以傳遞到所有 ab* 的 queue // headers : 不常用 $channel->exchange_declare('demo', 'direct'); //聲明交換機，相當于定義規則

?

//passive: 消極處理， 判斷是否存在隊列，存在則返回，不存在直接拋出 PhpAmqpLib\Exception\AMQPProtocolChannelException 異常 //durable：true、false true：在服務器重啟時，能夠存活 //exclusive ：是否為當前連接的專用隊列，在連接斷開后，會自動刪除該隊列，生產環境中應該很少用到吧。 //autodelete：當沒有任何消費者使用時，自動刪除該隊列 //nowait : 相當于一個async版的聲明隊列，可以看到是沒有返回的。調完方法就結束。也不等隊列創建結果 //auto_delete: 當所有消費客戶端連接斷開后，是否自動刪除隊列 //arguments: 附加參數，規則設置 $channel->queue_declare('kzae', false, false, false, false);

?

//隊列綁定交換機 $channel->queue_bind('hello','amq.direct', 'bbb');

?3.發送

//mandatory標志告訴服務器至少將該消息route到一個隊列中，否則將消息返還給生產者； //immediate標志告訴服務器如果該消息關聯的queue上有消費者，則馬上將消息投遞給它，如果所有queue都沒有消費者， //直接把消息返還給生產者，不用將消息入隊列等待消費者了 $channel->basic_publish($msg, 'amq.direct', 'bbb');

//通過 BasicQos 方法設置prefetchCount = 1。這樣RabbitMQ就會使得每個Consumer在同一個時間點最多處理一個Message
$channel->basic_qos(0, 1, false);

4.接收

//采用 no-ack 的方式進行確認，也就是說，每次 Consumer 接到數據后，不管是否處理完成，RabbitMQ Server 會立即把這個 Message 標記為完成，然后從 Queue 中刪除 $channel->basic_consume('hello', '', false, false, false, false, $callback);

?5.回調

$callback = function(PhpAmqpLib\Message\AMQPMessage $msg) {echo " [x] Received ", $msg->body, "\n";//delivery_info['channel']: PhpAmqpLib\Channel\AMQPChannelvar_dump($msg->delivery_info);$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']); //確認ack };

?

6. 結束

$channel->close(); $connection->close();

?

轉載于:https://www.cnblogs.com/yangs0/p/8451257.html

總結

以上是生活随笔為你收集整理的rabbitmq 一些基本函数介绍的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。