1，基本概念
producer：消息生产者，就是投递消息的程序
channel：通道，建立一个访问通道
Exchange：交换机，它指定消息按什么规则，路由到哪个队列
Binding:路由，exchange根据这个关键字(Routing Key)进行消息投递
Queue：队列，每个消息都会被投入到一个或多个队列
consumer：消息消费者，就是接受消息的程序
vhost：虚拟主机，开设多个vhost，用作不同用户的权限分离
2.安装rabbitMQ
3.下载php-amqplib扩展包
在你的项目目录下的composer.json文件中增加下面内容
{ "require": { "php-amqplib/php-amqplib": "2.7.*" //增加这行 }}
然后接着执行composer update php-amqplib/php-amqplib
4.建立生产者与mq之间的连接
$conn = new AMQPStreamConnection( $host,$port,$user,$password,$vhost);
$host: RabbitMQ服务器主机IP地址
$port: RabbitMQ服务器端口
$user: 连接RabbitMQ服务器的用户名
$password: 连接RabbitMQ服务器的用户密码
$vhost: 连接RabbitMQ服务器的vhost（虚拟主机）
5.建立通道
$channel = $conn->channel($channel_id); //在已连接基础上建立生产者与mq之间的通道
$channel_id 信道id，不传则获取$channel[“”]信道，再无则循环$this->channle数组，下标从1到最大信道数找第一个不是AMQPChannel对象的下标，实例化并返回AMQPChannel对象，无则抛出异常No free channel ids
6.声明初始化交换机
$channel->exchange_declare($exhcange_name,$type,$passive,$durable,$auto_delete); $exhcange_name 交换器名字
$type 交换器类型
$passive false 是否检测同名队列
$durable false 交换机是否开启持久化
$auto_detlete false 通道关闭后是否删除队列
交换器类型:direct(默认)，fanout, topic, 和headers
Direct:直接交换器，工作方式类似于单播，发送完全匹配路由键的队列
fanout:广播是式交换器，所以的队列都发送
topic:主题交换器，匹配模式相同的所有队列，（ * 表是匹配一个任意词组，#表示匹配0个或多个词组）
headers:消息体的header匹配（ignore）
7.声明初始化一条队列
$channel->queue_declare($name, $passive,$durable,$exclusive,$auto_delete);
$name: 队列名称
$passive: false 是否检测同名队列
$durable: false 是否开启队列持久化
$exclusive: false 队列是否可以被其他队列访问
$auto_delete: true 通道关闭后是否删除队列
8.路由绑定
$channel->queue_bind($queueName, $exchangeName, $routingKey);
$queueName 队列名称
$exchangeName 交换机名称
$routingKey 路由键
9.创建要发送的信息对象，可以创建多个消息
$msg = new AMQPMessage($data,$properties)
$data string类型 要发送的消息
$properties array类型 设置的属性，比如设置该消息持久化[‘delivery_mode’=>2]
10.发送消息
$channel->basic_publish($msg,$exchange,$routing_key);
$msg object AMQPMessage对象
$exchange string 交换机名字
$routing_key string 路由键 如果交换机类型
11.关闭信道和链接
$channel->close();
$connection->close();
12.设置消费者（Consumer）客户端同时只处理一条队列
$channel->basic_qos(0, 1, false);
13.消费消息
$channel->basic_consume($queue,$consumer_tag,$no_local, $no_ack, $exclusive,$nowait, $callback);
$queue队列名称
$consumer_tag 消费者客户端身份标识，用于区分多个客户端
$no_local 这个功能属于AMQP的标准，但是RabbitMQ并没有做实现
$no_ack 收到消息后，是否不需要回复确认即被认为被消费
$exclusive 是否排他，即这个队列只能由一个消费者消费。适用于任务不允许进行并发处理的情况下
$nowait 不返回执行结果，但是如果排他开启的话，则必须需要等待结果的，如果两个一起开就会报错
$callback回调函数
$callback = function($msg){
echo " [x] Received ", $msg->body, "\n";
sleep(substr_count($msg->body, '.'));
echo " [x] Done", "\n";
//手动确认ack，确保消息已经处理 需要开始$no_ack为true
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};
14.监听消息，一有消息，立马就处理
while(count($channel->callbacks)) {
$channel->wait();
}
15.六种工作模式
1，简单模式
<?php
require_once __DIR__ .'/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
$host = '127.0.0.1'; //ip
$port = '5672'; //端口
$username = 'test'; //用户名
$pwd = '123456'; //密码
//生产者
public function send($host,$port , $username,$pwd ) {
$connection = new AMQPStreamConnection($host, $port, $username, $pwd);
$channel = $connection->channel();
$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();
}
//消费者
public function consume($host, $port , $username, $pwd) {
$connection = new AMQPStreamConnection( $host , $port , $username , $pwd );
$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(); }
}
?>
