消息队列RabbitMQ入门与PHP实例详解

推荐学习：《PHP视频教程》

消息队列介绍以及消息队列应用场景

RabbitMQ

说明
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安装启动

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

php消息队列rabbitmq各种模式使用

rabbitmq管理界面及命令行使用

4369：epmd(Erlang Port Mapper Daemon)，erlang服务端口

5672 ：client端通信口

15672：HTTP API客户端，管理UI（仅在启用了管理插件的情况下）不一定会启动

25672：用于节点间通信（Erlang分发服务器端口）

rabbitmq 管理命令
启动15672：HTTP API客户端，管理UI（仅在启用了管理插件的情况下）

# 启动rabbitmq_management插件
rabbitmq-plugins  enable  rabbitmq_management

# 查看所有插件
rabbitmq-plugins  list

测试访问UI界面：（此时非localhost地址是无法登录）
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等。

1）查看不同用户的vhost

创建vhost，以及分配权限

# 新增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 ".*" ".*" ".*"
 ".*" ".*" ".*"后边三个.*分别代表：配置权限、写权限、读权限

2）为php安装rabbitmq扩展安装
https://github.com/php-amqplib/php-amqplib 扩展安装

修改阿里云镜像

composer config -g repo.packagist composer https://mirrors.aliyun.com/composer/

开始下载–这里有时候会下载成2.8低版本的，需要指定版本
，下载不成功则升级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队列可以进入到消息详情

simple模式消费者接受消息

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();

worker模式生产消费消息

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 < 10; $i++) {
    //生产数据
    $data = 'this is messge' . $i;
//创建消息
    $msg = new AMQPMessage($data, ['delivery_mode' => 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();

fanout模式生产者推送到交换器

发布/订阅模式
是要是公用一个交换机的消费端都能收到同样的消息，类似广播的功能

文档：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();

direct模式消息队列使用

文档：
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();

topic模式消息队列使用

通配符的匹配模式

如消费端中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视频教程》