介紹

RabbitMQ是一個(gè)消息代理器：它接受和轉(zhuǎn)發(fā)消息。你可以把它當(dāng)作一個(gè)郵局：當(dāng)你把郵件放在信箱里時(shí)，你可以肯定郵差先生最終會(huì)把郵件送到你的收件人那里。在這個(gè)比喻中，RabbitMQ就是這里的郵箱，郵局和郵差。

RabbitMQ和郵局之間的主要區(qū)別是，它不處理紙張，而是接受、存儲(chǔ)和轉(zhuǎn)發(fā)二進(jìn)制數(shù)據(jù)?消息。

RabbitMQ，和一般的消息傳遞，使用專業(yè)術(shù)語。

生產(chǎn)者的工作就是發(fā)送消息。發(fā)送消息的程序是生產(chǎn)者：

隊(duì)列類比一個(gè)郵箱，存在于RabbitMQ， 然而信息流通過RabbitMQ和您的應(yīng)用程序，他們只能存儲(chǔ)在一個(gè)隊(duì)列。隊(duì)列只受主機(jī)內(nèi)存和磁盤限制的約束，它本質(zhì)上是一個(gè)很大的消息緩沖區(qū)。會(huì)有許多生產(chǎn)者可以發(fā)送到一個(gè)隊(duì)列的消息，許多消費(fèi)者可以嘗試從一個(gè)隊(duì)列接收數(shù)據(jù)。這就是我們?nèi)绾伪硎娟?duì)列的方式：

消費(fèi)者和生產(chǎn)者有著相似的意義. 消費(fèi)者無非就是等待消息然后處理的程序:

請(qǐng)注意，生產(chǎn)者、消費(fèi)者和代理不必同一主機(jī)上；事實(shí)上，在大多數(shù)應(yīng)用程序中它們沒有這樣做。

"Hello World"

(使用PHP amqplib客戶端)

在本教程的這一部分中，我們將用PHP編寫兩個(gè)程序；一個(gè)生產(chǎn)者發(fā)送一條消息，一個(gè)用戶接收消息并將它們打印出來。我們會(huì)PHP amqplib API的忽略一些細(xì)節(jié)，集中在這個(gè)非常簡單的事情剛剛開始。這是一個(gè)“Hello World”的消息傳遞。

在下圖中，“p”是我們的生產(chǎn)商，“C”是我們的消費(fèi)者。在中間的框是一個(gè)隊(duì)列的消息緩沖區(qū)，RabbitMQ保持代表的消費(fèi)。

PHP amqplib客戶端庫

RabbitMQ有很多協(xié)議。本教程介紹AMQP 0-9-1，這是一個(gè)開放的、通用的協(xié)議消息。有許多不同的語言RabbitMQ一批客戶。我們將在本教程中使用PHP amqplib，composer解決依賴管理。

添加composer.json:

{

"require": {

"php-amqplib/php-amqplib": ">=2.6.1"

}

}

composer install

# 或者 直接運(yùn)行包引入

composer require php-amqplib/php-amqplib

現(xiàn)在我們可以開始我們的hello world

生產(chǎn)者(消息發(fā)送方)

我們命令我們的消息發(fā)布者(發(fā)送者)send.php和消息接收receive.php。發(fā)送者將連接到RabbitMQ，發(fā)送一條消息，然后退出。

require_once __DIR__ . '/vendor/autoload.php';

use PhpAmqpLib\Connection\AMQPStreamConnection;

use PhpAmqpLib\Message\AMQPMessage;

現(xiàn)在我們能創(chuàng)建一個(gè)連接服務(wù)器的Connection:

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');

$channel = $connection->channel();

該連接抽象套接字(socket)連接，并為我們負(fù)責(zé)協(xié)議版本協(xié)商和認(rèn)證等。這里，我們連接到一個(gè)rabbitmq代理器在本地機(jī)器上-使用localhost。如果我們想在不同的機(jī)器上連接到一個(gè)代理，我們只需在這里指定它的名稱或IP地址。

接下來，我們創(chuàng)建一個(gè)通道，這是處理事情的大部分API的地方。

發(fā)送消息前，我們必須聲明一個(gè)隊(duì)列為我們發(fā)送做準(zhǔn)備；然后我們可以向隊(duì)列發(fā)布消息：

$channel->queue_declare('hello', false, false, false, false);

$msg = new AMQPMessage('Hello World!');

$channel->basic_publish($msg, '', 'hello');

echo " [x] Sent 'Hello World!'\n";

聲明隊(duì)列是冪等的(原句：Declaring a queue is idempotent，這里的idempotent不知道是什么意思) - 只有在它不存在時(shí)才會(huì)創(chuàng)建隊(duì)列。消息內(nèi)容是一個(gè)字節(jié)數(shù)組，因此您可以在那里編碼用你喜歡的方式。

最后，我們關(guān)閉通道和連接；

$channel->close();

$connection->close();

上面我們完成了send.php.

接下來我們完成消費(fèi)方的代碼

消費(fèi)者(接收方，任務(wù)處理方)

消費(fèi)者從RabbitMQ接收推來的消息，我們會(huì)保持運(yùn)行監(jiān)聽消息并打印出來。

引入lib

require_once __DIR__ . '/vendor/autoload.php';

use PhpAmqpLib\Connection\AMQPStreamConnection;

設(shè)置與發(fā)布程序相同；我們打開一個(gè)連接和一個(gè)通道，并聲明將要消耗的隊(duì)列。注意，這與發(fā)送發(fā)布的隊(duì)列匹配。

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');

$channel = $connection->channel();

$channel->queue_declare('hello', false, false, false, false);

echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";

注意，我們也在這里聲明隊(duì)列。因?yàn)槲覀兛赡茉诎l(fā)布之前啟動(dòng)消費(fèi)者，我們希望在我們嘗試從它那里消費(fèi)消息之前確定隊(duì)列的存在。

我們將告訴服務(wù)器從隊(duì)列中發(fā)送消息。我們將定義一個(gè)PHP可調(diào)用，它將接收服務(wù)器發(fā)送的消息。請(qǐng)記住，消息是從服務(wù)器異步發(fā)送到客戶機(jī)的。

$callback = function($msg) {

echo " [x] Received ", $msg->body, "\n";

};

$channel->basic_consume('hello', '', false, true, false, false, $callback);

while(count($channel->callbacks)) {

$channel->wait();

}

當(dāng)調(diào)用basic_consume，我們的代碼會(huì)阻塞。當(dāng)我們收到消息時(shí)，我們的回調(diào)函數(shù)將通過接收到返回的消息傳遞。

以上是我們r(jià)eceive.php的代碼

運(yùn)行測(cè)試

運(yùn)行消費(fèi)者

php receive.php

運(yùn)行消息發(fā)送方

php send.php

列出隊(duì)列

rabbitmqctl list_queues

完整源碼(調(diào)整過)

config.php

return [

'vendor' => [

'path' => dirname(dirname(__DIR__)) . '/vendor'

],

'rabbitmq' => [

'host' => '127.0.0.1',

'port' => '5672',

'login' => 'qkl',

'password' => '123456',

'vhost' => '/'

]

];

?>

receive.php

$config = require "../config.php";

require_once $config['vendor']['path'] . '/autoload.php';

use PhpAmqpLib\Connection\AMQPStreamConnection;

use PhpAmqpLib\Message\AMQPMessage;

$connection = new AMQPStreamConnection($config['rabbitmq']['host'], $config['rabbitmq']['port'],

$config['rabbitmq']['login'], $config['rabbitmq']['password'], $config['rabbitmq']['vhost']);

$channel = $connection->channel();

$channel->queue_declare('hello', false, false, false, false);

echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";

$callback = function($msg) {

echo " [x] Received ", $msg->body, "\n";

};

$channel->basic_consume('hello', '', false, true, false, false, $callback);

while(count($channel->callbacks)) {

$channel->wait();

}

$channel->close();

$connection->close();

?>

send.php

$config = require "../config.php";

require_once $config['vendor']['path'] . '/autoload.php';

use PhpAmqpLib\Connection\AMQPStreamConnection;

use PhpAmqpLib\Message\AMQPMessage;

$connection = new AMQPStreamConnection($config['rabbitmq']['host'], $config['rabbitmq']['port'],

$config['rabbitmq']['login'], $config['rabbitmq']['password'], $config['rabbitmq']['vhost']);

$channel = $connection->channel();

//發(fā)送方其實(shí)不需要設(shè)置隊(duì)列， 不過對(duì)于持久化有關(guān)，建議執(zhí)行該行

$channel->queue_declare('hello', false, false, false, false);

$msg = new AMQPMessage('Hello World!');

$channel->basic_publish($msg, '', 'hello');

echo " [x] Sent 'Hello World!'\n";

$channel->close();

$connection->close();

?>

總結(jié)

以上是生活随笔為你收集整理的php rabbmq教程_RabbitMQ+PHP 教程一（Hello World）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。