PHP中RabbitMQ之phpAmqplib实现（五）

本章讲诉如何使用php-amqplib实现RabbitMQ。php

环境：CoentOS，PHP 7html

简单介绍一下php-amqpliblinux

php-amqplib是Advanced Message Queuing Protocol (AMQP)的一个PHP开源实现。高级消息队列协议（AMQP）是一个异步消息传递所使用的应用层协议规范。做为线路层协议，而不是API（例如JMS），AMQP 客户端可以无视消息的来源任意发送和接受信息json

一、RabbitMQ的安装centos

须要下载的两个包数组

erlang-21.0.7-1.el7.centos.x86_64.rpm服务器

rabbitmq-server-3.7.7-1.el7.noarch.rpm并发

这两个包我已经放在了百度云盘的分享上composer

连接：https://pan.baidu.com/s/1rMv_yFpLnH-D1S5wrOZrbA#list/path=%2FRabbitMQ异步

提取码：ipyu

而后参照 weixin_41368339的博客linux rabbitmq3.7.7安装与使用一文中的步骤安装步，基本上没有什么问题

二、composer的安装（已安装的请忽略此步）

为何要装这个？咱们能够经过composer来下载安装php-amqplib

如何安装composer，能够百度一下composer的全局安装或者直接去composer中文网

三、php-amqplib的下载及安装

新建一个composer.json的文件，内容以下所示

{ "require": { "php-amqplib/php-amqplib": ">=2.6.1" } }

而后执行

composer install

会生成一个composer.lock文件及vendor文件夹，vendor文件夹里有php-amqplib库，且有一个autoload.php文件可使用自动加载

四、Demo示例

本Demo示例只建立了一个直连交换机，共有四个文件Consumer.php （消费者），Publisher.php （生产者），Parenter.php （本身封装的RabbitMQ的方法），以及test.php （测试数据），目录如图所示

Parenter.php 代码以下图所示

<?php require_once __DIR__ . '/vendor/autoload.php'; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; abstract class Parenter { //MQ的默认链接配置
    public $config = array( '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->createConnect(); } //建立链接与信道
    private function createConnect() { $host = $this->config['host']; $port = $this->config['port']; $user = $this->config['user']; $password = $this->config['password']; $vhost = $this->config['vhost']; if(empty($host) || empty($port) || empty($user) || empty($password)) { throw new Exception('RabbitMQ的链接配置不正确'); } //建立连接
        $this->connection = new AMQPStreamConnection($host, $port, $user, $password, $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, array('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; } } }

Consumer.php （消费者）

<?php include_once('Parenter.php'); class Consumer extends Parenter { public function __construct() { parent::__construct('exchange', 'queue', 'routeKey'); } public function doProcess($msg) { echo $msg."\n"; } } $consumer = new Consumer(); //$consumer->dealMq(false);
$consumer->dealMq(true);

Publisher.php （生产者）

<?php include_once('Parenter.php'); class Publisher extends Parenter { public function __construct() { parent::__construct('exchange', '', 'routeKey'); } public function doProcess($msg) { } }

test.php（测试数据）

五、添加交换机与队列

打开http://ip(你的RabbitMQ安装的主机):15672/，会进入到RabbitMQ的可视化管理后台登陆页面，登陆你的帐号密码（若是你是按照第一步提到的博客里的教程来装的，那你的帐号密码就是guest），而后新加交换机和队列，

如下是新加交换机的操做，注意vhost与以及交换机的名称要与代码里的消费者与生产者传入的参数值保持一致，若是你不想使用"/"这个默认的vhost，也能够新建一个vhost（什么？你问我如何新建，那么请百度一下），可是要记住在代码里建立消费者与生产者时把你新加的这个vhost传进去，覆盖RabbitMqParernt.php里的vhost

如下是新加队列，这里的vhost要与上一步的vhost保持一致，保证交换机与队列在同一个vhost下，否则交换机会找不到队列的，队列名与消费者代码里传入进去的队列名保持一致

六、运行代码

先打开一个窗口启动消费者

运行测试脚本

若是打印出来字符串就成功了

注意：消费者与生产者传入的交换机名称，路由键必须相同

交换机类型请务必选择直连，各类交换机的路由键形式不大相同，有兴趣的同窗能够去试试其它类型的交换机实现哦

当修改了vhost或者交换机名称，队列名称等时，须要修改对应代码

至于注释里的ack应答，我会在以后的博客里详细介绍，包括RabbitMQ的持久化，这里使用默认的ack应答便可

代码里不少注释都是我后来学习php-amqplib库中类的方法时加的，表示的是参数的意义，你们也能够去研究一下，这里提供个网址： Rabbitmq各方法的做用详解

关于管理后台及RabbitMQ的命令，我这里就很少介绍了，有兴趣的同窗去网上搜索一下就能搜到好多

【转】：https://blog.csdn.net/yeyun666/article/details/86743784