這篇文章為大家帶來了關於PHP的相關知識，其中主要介紹了關於消息隊列RabbitMQ入門還有一些實戰詳解，消息隊列是應用間的通信方式，下面一起來看一下，希望對大家有幫助。
推薦學習：《PHP影片教學》
說明
MQ(Message Queue) 即訊息佇列，是應用程式間的通訊方式，訊息發送後可立即返回，由訊息系統來確保訊息的可靠傳遞。 」訊息佇列「是在訊息的傳輸過程中保存訊息的容器。它是典型的：生產者、消費者模式。生產者不斷向訊息隊列中生產訊息，消費者不斷的從隊列中獲取訊息。因為訊息的生產和消費都是非同步的，而且只關心訊息的發送和接收，沒有業務邏輯的侵入，這樣就實現生產者和消費者的解耦。
為什麼要使用訊息中間件？
訊息佇列是分散式系統中重要的元件，解決應用解耦，非同步訊息，流量削峰等問題，實現高並發，高可用，可伸縮和最終一致性架構
異步處理
用戶註冊資訊後需要發送郵件和註冊短信
1、用戶註冊信息寫入數據庫後即使返回註冊成功的信息
2、發送郵件和註冊短信透過訊息佇列非同步執行，使用者不需要等待這兩個動作
應用程式解耦
使用者下單後，訂單系統需要通知庫存系統。傳統的做法是，訂單系統調用庫存系統的接口，進行增減庫存
1、用戶下單入列生產，返回成功提示
2、隊列消費庫存系統，進行庫存增減
流量削峰
流量削峰也是訊息佇列中的常見場景，一般在秒殺或團搶活動中使用廣泛
1、當一批用戶請求過來進入列隊，控制入列數量，超出一定數量返回秒殺結束
2、然後隊列一個個按照先進先出進行隊列消費
Rabbitmq特性
可靠性（Reliability） RabbitMQ 使用一些機制來保證可靠性，如持久化、傳輸確認、發布確認。
靈活的路由（Flexible Routing） 在訊息進入佇列之前，透過 Exchange 來路由訊息的。對於典型的路由功能，RabbitMQ 已經提供了一些內建的 Exchange 來實現。針對更複雜的路由功能，可以將多個 Exchange 綁定在一起，也透過插件機制實現自己的 Exchange 。
訊息集群（Clustering） 多個 RabbitMQ 伺服器可以組成一個集群，形成一個邏輯 Broker 。
高可用（Highly Available Queues） 佇列可以在叢集中的機器上進行鏡像，使得在部分節點出問題的情況下佇列仍然可用。
多種協定（Multi-protocol） RabbitMQ 支援多種訊息佇列協議，例如 STOMP、MQTT 等等。
多語言客戶端（Many Clients） RabbitMQ 幾乎支援所有常用語言，例如PHP Java、.NET、Ruby 等等。
管理介面（Management UI） RabbitMQ 提供了一個易用的使用者介面，使得使用者可以監控和管理訊息 Broker 的許多方面。
追蹤機制（Tracing） 如果訊息異常，RabbitMQ 提供了訊息追蹤機制，使用者可以找出發生了什麼事。
插件機制（Plugin System） RabbitMQ 提供了許多插件，來從多方面進行擴展，也可以編寫自己的插件。
RabbitMQ的工作原理
#Broker: 接收和分發訊息的應用，RabbitMQ Server就是Message Broker。
Virtual host: 類似mysql的資料庫，當多個不同的使用者使用同一個RabbitMQ server提供的服務時，可以分割出多個vhost，每個使用者在自己的vhost創作exchange／queue等。
Connection: publisher／consumer與broker之間的TCP連結。
Channel: 如果每一次造訪RabbitMQ都建立一個Connection，在訊息量大的時候建立TCP Connection的開銷將是巨大的，效率也較低。 Channel是在connection內部建立的邏輯連線Channel作為輕量級的Connection極大減少了作業系統建立TCP connection的開銷。
Exchange: message到達broker的第一站，根據分發規則，匹配查詢表中的routing key，分發訊息到queue中去。常用的類型有：direct (point-to-point), topic (publish-subscribe) and fanout (multicast)。
Queue: 訊息最後被送到這裡等待consumer取走。一個message可以同時拷貝到多個queue中。
RabbitMQ官方網址：http://www.rabbitmq.com
安裝rabbitmq需要先安裝erlang
第一步： erlang 安裝
安裝rabbitmq需要先安裝erlang，centos7不支援erlang 24版本的安裝
下載：
# 系统 centos 7# 下载erlang包，手动下载后上传至服务器，我在使用wget下载后无法安装，这里没明白 # 安装 yum install erlang-23.3.4.4-1.el7.x86_64.rpm # 验证安装是否成功 erl
第二步驟：安裝rabbitmq
# 系统 centos 7# 下载rabbitmq包，手动下载后上传至服务器，我在使用wget下载后无法安装，这里没明白 # 安装 yum install rabbitmq-server-3.8.19-1.el7.noarch.rpm # 启动 systemctl start rabbitmq-server # 关闭 systemctl stop rabbitmq-server # 查看默认端口服务是否启动 netstat -tunlp
rabbitmq 管理指令 啟動15672：HTTP API客戶端，管理UI（僅在啟用了管理外掛程式的情況下）
# 启动rabbitmq_management插件 rabbitmq-plugins enable rabbitmq_management # 查看所有插件 rabbitmq-plugins list
http://192.168.10.105:15672/
rabbitmq 設定管理介面#
# 新增一个用户 rabbitmqctl add_user 【用户名Username】 【密码Password】 rabbitmqctl add_user root root # 删除一个用户 rabbitmqctl delete_user Username # 修改用户的密码 rabbitmqctl change_password Username Newpassword # 查看当前用户列表 rabbitmqctl list_users # 设置用户角色的命令为： rabbitmqctl set_user_tags User Tag rabbitmqctl set_user_tags root administrator # User为用户名， Tag为角色名(对应于上面的administrator，monitoring，policymaker，management，或其他自定义名称)。
#命令列建立vhost以及php擴充安裝 類似mysql的資料庫，當多個不同的使用者使用同一個RabbitMQ server提供的服務時，可以分割出多個vhost，每個使用者在自己的vhost建立exchange/queue等。
# 新增vhost rabbitmqctl add_vhost vhostname rabbitmqctl add_vhost order # 查看vhost列表 rabbitmqctl list_vhosts #为vhost添加用户 rabbitmqctl set_permissions -p vhostname username ".*" ".*" ".*"rabbitmqctl set_permissions -p order root ".*" ".*" ".*" ".*" ".*" ".*"后边三个.*分别代表：配置权限、写权限、读权限
https://github.com/php-amqplib/php-amqplib 擴充安裝
composer config -g repo.packagist composer https://mirrors.aliyun.com/composer/
，下載不成功則升級composer、php.ini 打開sockets 擴展和切換國內鏡像
# 升级composer composer self-update #php.ini 打开 sockets 扩展 #下载指定版本 composer require php-amqplib/php-amqplib=^3.0
simple模式生產者訊息推送到訊息佇列 文件：
https://www.rabbitmq.com/tutorials/tutorial-one-php.html
生產者代碼
http://localhost/rabbitmq/simple/pro.php
<?php require_once "../vendor/autoload.php"; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; //生产者 //Connection: publisher／consumer和broker之间的TCP连接 //Channel: 如果每一次访问RabbitMQ都建立一个Connection，在消息量大的时候建立TCP Connection的开销将是巨大的，效率也较低。Channel是在connection内部建立的逻辑连接Channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销。 //建立connction $connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order'); //Channel $channel = $connection->channel(); //声明队列名为：goods $queue_name = 'goods'; $channel->queue_declare($queue_name, false, true, false, false); //生产数据 $data = 'this is messge'; //创建消息 $msg = new AMQPMessage($data, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_NON_PERSISTENT]); //发布消息 $channel->basic_publish($msg, $exchange = '', $queue_name); //关闭连接 $channel->close(); $connection->close();
运行生产者脚本：
http://localhost/rabbitmq/simple/pro.php
点击goods队列可以进入到消息详情
http://localhost/rabbitmq/simple/con.php
<?php require_once "../vendor/autoload.php";
use PhpAmqpLib\Connection\AMQPStreamConnection;
//建立connction
$connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order');
//Channel
$channel = $connection->channel();
//声明队列名为：goods
$queue_name = 'goods';
$channel->queue_declare($queue_name, false, true, false, false);
echo " [*] Waiting for messages. To exit press CTRL+C\n";
$callback = function ($msg) {
echo 'received = ', $msg->body . "\n";
};
//开启消费
$channel->basic_consume($queue_name, '', false, true, false, false, $callback);
//不断的循环进行消费
while ($channel->is_open()) {
$channel->wait();
}
//关闭连接
$channel->close();
$connection->close();rabbitmq Work Queues
一个生产者对应多个消费者，消费特别慢时增加几个消费分发
生产者，和上文生产者不变
<?php require_once "../vendor/autoload.php"; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; //生产者 //Connection: publisher／consumer和broker之间的TCP连接 //Channel: 如果每一次访问RabbitMQ都建立一个Connection，在消息量大的时候建立TCP Connection的开销将是巨大的，效率也较低。Channel是在connection内部建立的逻辑连接Channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销。 //建立connction $connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order'); //Channel $channel = $connection->channel(); //声明队列名为：task_queue $queue_name = 'task_queue'; $channel->queue_declare($queue_name, false, true, false, false); for ($i = 0; $i AMQPMessage::DELIVERY_MODE_NON_PERSISTENT]); //发布消息 $channel->basic_publish($msg, $exchange = '', $queue_name); } //关闭连接 $channel->close(); $connection->close();
消费者worker1
D:\phpstudy_pro\WWW\rabbitmq\worker\worker1.php
<?php require_once "../vendor/autoload.php";
use PhpAmqpLib\Connection\AMQPStreamConnection;
//建立connction
$connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order');
//Channel
$channel = $connection->channel();
//声明队列名为：task_queue
$queue_name = 'task_queue';
$channel->queue_declare($queue_name, false, true, false, false);
echo " [*] Waiting for messages. To exit press CTRL+C\n";
$callback = function ($msg) {
echo 'received = ', $msg->body . "\n";
};
//开启消费
$channel->basic_consume($queue_name, '', false, true, false, false, $callback);
//不断的循环进行消费
while ($channel->is_open()) {
$channel->wait();
}
//关闭连接
$channel->close();
$connection->close();消费者worker2，代码和worker1一样，同时运行开启后会一起消费
D:\phpstudy_pro\WWW\rabbitmq\worker\worker2.php
消费者消费消息ack确认
用以确认不会丢失消息
消费消息
basic_consume($queue = ‘’, $consumer_tag = ‘’, $no_local = false, $no_ack = false, $exclusive = false, $nowait = false, $callback = null, $ticket = null, $arguments = array())
no_ack=false,设置为手动应答
开启后需要进行消息的消费确认后才会进行移除，否者该消息会一直存在消息队列中
消费端代码
D:\phpstudy_pro\WWW\rabbitmq\worker\worker1.php
<?php require_once "../vendor/autoload.php";
use PhpAmqpLib\Connection\AMQPStreamConnection;
//建立connction
$connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order');
//Channel
$channel = $connection->channel();
//声明队列名为：task_queue
$queue_name = 'task_queue';
$channel->queue_declare($queue_name, false, true, false, false);
echo " [*] Waiting for messages. To exit press CTRL+C\n";
$callback = function ($msg) {
echo 'received = ', $msg->body . "\n";
//确认消息已被消费，从生产队列中移除
$msg->ack();
};
//设置消费成功后才能继续进行下一个消费
$channel->basic_qos(null, 1, null);
//开启消费no_ack=false,设置为手动应答
$channel->basic_consume($queue_name, '', false, false, false, false, $callback);
//不断的循环进行消费
while ($channel->is_open()) {
$channel->wait();
}
//关闭连接
$channel->close();
$connection->close();发布/订阅模式
是要是公用一个交换机的消费端都能收到同样的消息，类似广播的功能
文档：rabbitmq Publish/Subscribe
https://www.rabbitmq.com/tutorials/tutorial-three-php.html
rabbitmq Exchange类型
交换器、路由键、绑定 Exchange：交换器。发送消息的AMQP实体。交换器拿到一个消息之后将它路由给一个或几个队列。它使用哪种路由算法是由交换机类型和被称作绑定（Binding）的规则所决定的。RabbitMQ有四种类型。 RoutingKey：路由键。生产者将消息发送给交换器。一般会指定一个RoutingKey,用来指定这个消息的路由规则，而这个RoutingKey需要与交换器类型和绑定键（BindingKey）联合使用才能最终失效。 Binding：绑定。绑定（Binding）是交换机（Exchange）将消息（Message）路由给队列（Queue）所需遵循的规则。 # 四种模式 Direct 定向 消息与一个特定的路由键完全匹配 Topic 通配符 路由键和某模式进行匹配 Fanout 广播 发送到该类型交换机的消息都会被广播到与该交换机绑定的所有队列 Headers 不处理路由键，而是根据发送的消息内容中的headers属性进行匹配
exchange_declare($exchange, $type, $passive = false, $durable = false, $auto_delete = true, $internal = false, $nowait = false, $arguments = array(), $ticket = null) 。试探性申请一个交换器，若该交换器不存在，则创建；若存在，则跳过。
生产者代码
D:\phpstudy_pro\WWW\rabbitmq\ps\pro.php
<?php require_once "../vendor/autoload.php"; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; //建立connction $connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order'); //Channel $channel = $connection->channel(); //声明交换器 $exc_name = 'exch'; $channel->exchange_declare($exc_name, 'fanout', false, false, false); //声明数据 $data = 'this is fanout message'; //创建消息 $msg = new AMQPMessage($data, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_NON_PERSISTENT]); //发布消息 $channel->basic_publish($msg, $exc_name); //关闭连接 $channel->close(); $connection->close();
fanout模式消费者消费消息
是要是公用一个交换机的消费端都能收到同样的消息，类似广播的功能
当消费端运行时才会显示该队列
消费端：
D:\phpstudy_pro\WWW\rabbitmq\ps\worker1.php
<?php require_once "../vendor/autoload.php";
use PhpAmqpLib\Connection\AMQPStreamConnection;
//建立connction
$connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order');
//Channel
$channel = $connection->channel();
//声明交换器
$exc_name = 'exch';
$channel->exchange_declare($exc_name, 'fanout', false, false, false);
//获取系统生成的消息队列名称
list($queue_name, ,) = $channel->queue_declare('', false, false, true, false);
//将队列名与交换器名进行绑定
$channel->queue_bind($queue_name,$exc_name);
$callback = function ($msg) {
echo 'received = ', $msg->body . "\n";
//确认消息已被消费，从生产队列中移除
$msg->ack();
};
//设置消费成功后才能继续进行下一个消费
$channel->basic_qos(null, 1, null);
//开启消费no_ack=false,设置为手动应答
$channel->basic_consume($queue_name, '', false, false, false, false, $callback);
//不断的循环进行消费
while ($channel->is_open()) {
$channel->wait();
}
//关闭连接
$channel->close();
$connection->close();文档：
https://www.rabbitmq.com/tutorials/tutorial-four-php.html
用来指定不同的交换机和指定routing_key，在消费端进行消费
生产者代码：
D:\phpstudy_pro\WWW\rabbitmq\routing\pro.php
<?php require_once "../vendor/autoload.php"; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; //建立connction $connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order'); //Channel $channel = $connection->channel(); //声明交换器 $exc_name = 'direct_log'; //指定routing_key $routing_key = 'info'; //指定交换机类型为direct $channel->exchange_declare($exc_name, 'direct', false, false, false); //声明数据 $data = 'this is ' . $routing_key . ' message'; //创建消息 $msg = new AMQPMessage($data, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_NON_PERSISTENT]); //发布消息 //指定使用的routing_key $channel->basic_publish($msg, $exc_name, $routing_key); //关闭连接 $channel->close(); $connection->close();
消费者代码
D:\phpstudy_pro\WWW\rabbitmq\routing\info.php
<?php require_once "../vendor/autoload.php";
use PhpAmqpLib\Connection\AMQPStreamConnection;
//建立connction
$connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order');
//Channel
$channel = $connection->channel();
//声明交换器
$exc_name = 'direct_log';
//指定routing_key
$routing_key = 'info';
$channel->exchange_declare($exc_name, 'direct', false, false, false);
//获取系统生成的消息队列名称
list($queue_name, ,) = $channel->queue_declare('', false, false, true, false);
//将队列名与交换器名进行绑定，并指定routing_key
$channel->queue_bind($queue_name,$exc_name,$routing_key);
$callback = function ($msg) {
echo 'received = ', $msg->body . "\n";
//确认消息已被消费，从生产队列中移除
$msg->ack();
};
//设置消费成功后才能继续进行下一个消费
$channel->basic_qos(null, 1, null);
//开启消费no_ack=false,设置为手动应答
$channel->basic_consume($queue_name, '', false, false, false, false, $callback);
//不断的循环进行消费
while ($channel->is_open()) {
$channel->wait();
}
//关闭连接
$channel->close();
$connection->close();通配符的匹配模式
如消费端中routing_key = ‘user.*’;
生产者：
指定routing_key= ‘user.top’
<?php require_once "../vendor/autoload.php"; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; //建立connction $connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order'); //Channel $channel = $connection->channel(); //声明交换器 $exc_name = 'topic_log'; //指定routing_key $routing_key = 'user.top'; //指定交换机类型为direct $channel->exchange_declare($exc_name, 'topic', false, false, false); //声明数据 $data = 'this is ' . $routing_key . ' message'; //创建消息 $msg = new AMQPMessage($data, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_NON_PERSISTENT]); //发布消息 //指定使用的routing_key $channel->basic_publish($msg, $exc_name, $routing_key); //关闭连接 $channel->close(); $connection->close();
消费者
消费端中routing_key = ‘user.*’;
<?php require_once "../vendor/autoload.php";
use PhpAmqpLib\Connection\AMQPStreamConnection;
//建立connction
$connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order');
//Channel
$channel = $connection->channel();
//声明交换器
$exc_name = 'direct_log';
//指定routing_key
$routing_key = 'user.*';
$channel->exchange_declare($exc_name, 'topic', false, false, false);
//获取系统生成的消息队列名称
list($queue_name, ,) = $channel->queue_declare('', false, false, true, false);
//将队列名与交换器名进行绑定，并指定routing_key
$channel->queue_bind($queue_name,$exc_name,$routing_key);
$callback = function ($msg) {
echo 'received = ', $msg->body . "\n";
//确认消息已被消费，从生产队列中移除
$msg->ack();
};
//设置消费成功后才能继续进行下一个消费
$channel->basic_qos(null, 1, null);
//开启消费no_ack=false,设置为手动应答
$channel->basic_consume($queue_name, '', false, false, false, false, $callback);
//不断的循环进行消费
while ($channel->is_open()) {
$channel->wait();
}
//关闭连接
$channel->close();
$connection->close();推荐学习：《PHP视频教程》
以上是訊息佇列RabbitMQ入門與PHP實例詳解的詳細內容。更多資訊請關注PHP中文網其他相關文章！
