地点：https://github.com/php-amqplib/php-amqplib

composer require php-amqplib/php-amqplib

新建 RabbitMq.php

<?php require "vendor/autoload.php"; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; abstract class RabbitMq { public $config = [ ‘host‘ => ‘127.0.0.1‘, //ip ‘port‘ => 5672, //端标语 ‘user‘ => ‘guest‘, //用户 ‘password‘ => ‘guest‘, //暗码 ‘vhost‘ => ‘/‘ //虚拟host ]; public $connection; //链接 public $channel; //信道 public $exchangeName = ‘‘; //交换机名 public $queueName = ‘‘; //行列队伍名 public $routeKey = ‘‘; //路由键 public $exchangeType = ‘direct‘; //交换机类型 public $autoAck = true; //是否自动ack应答 //初始化 public function __construct($exchangeName, $queueName, $routeKey, $exchangeType = ‘direct‘, $config = array()) { $this->exchangeName = empty($exchangeName) ? ‘‘ : $exchangeName; $this->queueName = empty($queueName) ? ‘‘ : $queueName; $this->routeKey = empty($routeKey) ? ‘‘ : $routeKey; $this->exchangeType = empty($exchangeType) ? ‘‘ : ‘direct‘; if (!empty($config)) { $this->setConfig($config); } //创建链接 $this->connection = new AMQPStreamConnection($this->config[‘host‘], $this->config[‘port‘], $this->config[‘user‘], $this->config[‘password‘], $this->config[‘vhost‘]); //创建信道 $this->channel = $this->connection->channel(); $this->createExchange(); } //创建交换机 private function createExchange() { //创建交换机$channel->exchange_declare($exhcange_name,$type,$passive,$durable,$auto_delete); //passive: 消极措置惩罚惩罚， 判断是否存在行列队伍，存在则返回，不存在直接抛出 PhpAmqpLib\Exception\AMQPProtocolChannelException 异常 //durable：true、false true：处事器重启会保存下来Exchange。警告：仅设置此选项，不代表动静长期化。即不保证重启后动静还在 //autoDelete:true、false.true:当已经没有消费者时，处事器是否可以删除该Exchange $this->channel->exchange_declare($this->exchangeName, $this->exchangeType, false, true, false); //passive: 消极措置惩罚惩罚， 判断是否存在行列队伍，存在则返回，不存在直接抛出 PhpAmqpLib\Exception\AMQPProtocolChannelException 异常 //durable：true、false true：在处事器重启时，能够存活 //exclusive ：是否为当前连接的专用行列队伍，在连接断开后，会自动删除该行列队伍 //autodelete：当没有任何消费者使用时，自动删除该行列队伍 //arguments: 自界说法则 $this->channel->queue_declare($this->queueName, false, true, false, false); } //发送动静 public function sendMessage($data) { //创建动静$msg = new AMQPMessage($data,$properties) //#$data string类型 要发送的动静 //#roperties array类型 设置的属性，，好比设置该动静长期化[‘delivery_mode’=>2] $msg = new AMQPMessage($data, [‘delivery_mode‘ => AMQPMessage::DELIVERY_MODE_PERSISTENT]); $this->channel->basic_publish($msg,$this->exchangeName, $this->routeKey); } //措置惩罚惩罚动静 public function dealMq($flag) { $this->autoAck = $flag; $this->channel->queue_bind($this->queueName,$this->exchangeName, $this->routeKey); //prefetchSize：0 //prefetchCount：会报告RabbitMQ不要同时给一个消费者推送多于N个动静，即一旦有N个动静还没有ack，则该consumer将block失，直到有动静ack //global：true\false 是否将上面设置应用于channel，简单点说，就是上面限制是channel级另外还是consumer级别 //$this->channel->basic_qos(0, 1, false); //1:queue 要取得动静的行列队伍名 //2:consumer_tag 消费者标签 //3:no_local false这个成果属于AMQP的标准,但是rabbitMQ并没有做实现.参考 //4:no_ack false收到动静后,是否不需要答复确认即被认为被消费 //5:exclusive false排他消费者,即这个行列队伍只能由一个消费者消费.适用于任务不允许进行并发措置惩罚惩罚的情况下.好比系统对接 //6:nowait false不返回执行功效,但是如果排他开启的话,则必需需要期待功效的,如果两个一起开就会报错 //7:callback null回调函数 //8:ticket null //9:arguments null $this->channel->basic_consume($this->queueName, ‘‘, false, $this->autoAck, false, false, function($msg){$this->get($msg);}); //监听动静 while(count($this->channel->callbacks)){ $this->channel->wait(); } } public function get($msg) { $param = $msg->body; $this->doProcess($param); if(!$this->autoAck) { //手动ack应答 $msg->delivery_info[‘channel‘]->basic_ack($msg->delivery_info[‘delivery_tag‘]); } } abstract public function doProcess($param); public function closeConnetct() { $this->channel->close(); $this->connection->close(); } //从头设置MQ的链接配置 public function setConfig($config) { if (!is_array($config)) { throw new Exception(‘config不是一个数组‘); } foreach ($config as $key => $value) { $this->config[$key] = $value; } } }

2.在创建 Consumer.php 消费