PHP中RabbitMQ之phpAmqplib实现（五

Posted 2022-06-07 php-linux

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

环境：CoentOS，PHP 7

简单介绍一下php-amqplib

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

1、RabbitMQ的安装

需要下载的两个包

erlang-21.0.7-1.el7.centos.x86_64.rpm

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

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

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

提取码：ipyu

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

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

为什么要装这个？我们可以通过composer来下载安装php-amqplib

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

3、php-amqplib的下载及安装

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

1.  
2.  
   "require":
3.  
   "php-amqplib/php-amqplib": ">=2.6.1"
4.  
5.  

然后执行

composer install

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

4、Demo示例

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

Parenter.php 代码如下图所示

1.  
   <?php
2.  
   require_once __DIR__ . ‘/vendor/autoload.php‘;
3.  
    
4.  
   use PhpAmqpLib\Connection\AMQPStreamConnection;
5.  
   use PhpAmqpLib\Message\AMQPMessage;
6.  
   abstract class Parenter
7.  
8.  
   //MQ的默认连接配置
9.  
   public $config = array(
10.  
    ‘host‘ => ‘127.0.0.1‘, //ip
11.  
    ‘port‘ => ‘5672‘, //端口号
12.  
    ‘user‘ => ‘guest‘, //用户
13.  
    ‘password‘ => ‘guest‘, //密码
14.  
    ‘vhost‘ => ‘/‘ //虚拟host
15.  
    );
16.  
     
17.  
    public $connection; //链接
18.  
    public $channel; //信道
19.  
     
20.  
    public $exchangeName = ‘‘; //交换机名
21.  
    public $queueName = ‘‘; //队列名
22.  
    public $routeKey = ‘‘; //路由键
23.  
    public $exchangeType = ‘direct‘; //交换机类型
24.  
     
25.  
    public $autoAck = true; //是否自动ack应答
26.  
     
27.  
    public function __construct($exchangeName, $queueName, $routeKey, $exchangeType = ‘direct‘, $config=array())
28.  
29.  
    $this->exchangeName = empty($exchangeName) ? ‘‘ : $exchangeName;
30.  
    $this->queueName = empty($queueName) ? ‘‘ : $queueName;
31.  
    $this->routeKey = empty($routeKey) ? ‘‘ : $routeKey;
32.  
    $this->exchangeType = empty($exchangeType) ? ‘‘ : ‘direct‘;
33.  
    if(!empty($config))
34.  
35.  
    $this->setConfig($config);
36.  
37.  
    $this->createConnect();
38.  
39.  
     
40.  
    //创建连接与信道
41.  
    private function createConnect()
42.  
43.  
    $host = $this->config[‘host‘];
44.  
    $port = $this->config[‘port‘];
45.  
    $user = $this->config[‘user‘];
46.  
    $password = $this->config[‘password‘];
47.  
    $vhost = $this->config[‘vhost‘];
48.  
    if(empty($host) || empty($port) || empty($user) || empty($password))
49.  
50.  
    throw new Exception(‘RabbitMQ的连接配置不正确‘);
51.  
52.  
    //创建链接
53.  
    $this->connection = new AMQPStreamConnection($host, $port, $user, $password, $vhost);
54.  
    //创建信道
55.  
    $this->channel = $this->connection->channel();
56.  
    $this->createExchange();
57.  
58.  
     
59.  
    //创建交换机
60.  
    private function createExchange()
61.  
62.  
    //创建交换机$channel->exchange_declare($exhcange_name,$type,$passive,$durable,$auto_delete);
63.  
    //passive: 消极处理， 判断是否存在队列，存在则返回，不存在直接抛出 PhpAmqpLib\Exception\AMQPProtocolChannelException 异常
64.  
    //durable：true、false true：服务器重启会保留下来Exchange。警告：仅设置此选项，不代表消息持久化。即不保证重启后消息还在
65.  
    //autoDelete:true、false.true:当已经没有消费者时，服务器是否可以删除该Exchange
66.  
    $this->channel->exchange_declare($this->exchangeName, $this->exchangeType, false, true, false);
67.  
    //passive: 消极处理， 判断是否存在队列，存在则返回，不存在直接抛出 PhpAmqpLib\Exception\AMQPProtocolChannelException 异常
68.  
    //durable：true、false true：在服务器重启时，能够存活
69.  
    //exclusive ：是否为当前连接的专用队列，在连接断开后，会自动删除该队列
70.  
    //autodelete：当没有任何消费者使用时，自动删除该队列
71.  
    //arguments: 自定义规则
72.  
    $this->channel->queue_declare($this->queueName, false, true, false, false);
73.  
74.  
     
75.  
    //发送消息
76.  
    public function sendMessage($data)
77.  
78.  
    //创建消息$msg = new AMQPMessage($data,$properties)
79.  
    //#$data string类型 要发送的消息
80.  
    //#roperties array类型 设置的属性，比如设置该消息持久化[‘delivery_mode’=>2]
81.  
    $msg = new AMQPMessage($data, array(‘delivery_mode‘ => AMQPMessage::DELIVERY_MODE_PERSISTENT));
82.  
    $this->channel->basic_publish($msg,$this->exchangeName, $this->routeKey);
83.  
84.  
     
85.  
    //处理消息
86.  
    public function dealMq($flag)
87.  
88.  
    $this->autoAck = $flag;
89.  
    $this->channel->queue_bind($this->queueName,$this->exchangeName, $this->routeKey);
90.  
    //prefetchSize：0
91.  
    //prefetchCount：会告诉RabbitMQ不要同时给一个消费者推送多于N个消息，即一旦有N个消息还没有ack，则该consumer将block掉，直到有消息ack
92.  
    //global：true\false 是否将上面设置应用于channel，简单点说，就是上面限制是channel级别的还是consumer级别
93.  
    //$this->channel->basic_qos(0, 1, false);
94.  
    //1:queue 要取得消息的队列名
95.  
    //2:consumer_tag 消费者标签
96.  
    //3:no_local false这个功能属于AMQP的标准,但是rabbitMQ并没有做实现.参考
97.  
    //4:no_ack false收到消息后,是否不需要回复确认即被认为被消费
98.  
    //5:exclusive false排他消费者,即这个队列只能由一个消费者消费.适用于任务不允许进行并发处理的情况下.比如系统对接
99.  
    //6:nowait false不返回执行结果,但是如果排他开启的话,则必须需要等待结果的,如果两个一起开就会报错
100.  
     //7:callback null回调函数
101.  
     //8:ticket null
102.  
     //9:arguments null
103.  
     $this->channel->basic_consume($this->queueName, ‘‘, false, $this->autoAck, false, false, function($msg)$this->get($msg););
104.  
     //监听消息
105.  
     while(count($this->channel->callbacks))
106.  
     $this->channel->wait();
107.  
108.  
109.  
      
110.  
     public function get($msg)
111.  
112.  
     $param = $msg->body;
113.  
     $this->doProcess($param);
114.  
     if(!$this->autoAck)
115.  
116.  
     //手动ack应答
117.  
     $msg->delivery_info[‘channel‘]->basic_ack($msg->delivery_info[‘delivery_tag‘]);
118.  
119.  
120.  
      
121.  
     abstract public function doProcess($param);
122.  
      
123.  
     public function closeConnetct()
124.  
125.  
     $this->channel->close();
126.  
     $this->connection->close();
127.  
128.  
      
129.  
     //重新设置MQ的链接配置
130.  
     public function setConfig($config)
131.  
132.  
     if (!is_array($config))
133.  
134.  
     throw new Exception(‘config不是一个数组‘);
135.  
136.  
     foreach($config as $key => $value)
137.  
138.  
     $this->config[$key] = $value;
139.  
140.  
141.  

Consumer.php （消费者）

1.  
   <?php
2.  
   include_once(‘Parenter.php‘);
3.  
   class Consumer extends Parenter
4.  
5.  
   public function __construct()
6.  
7.  
   parent::__construct(‘exchange‘, ‘queue‘, ‘routeKey‘);
8.  
9.  
   public function doProcess($msg)
10.  
11.  
    echo $msg."\n";
12.  
13.  
14.  
    $consumer = new Consumer();
15.  
    //$consumer->dealMq(false);
16.  
    $consumer->dealMq(true);

Publisher.php （生产者）

1.  
   <?php
2.  
   include_once(‘Parenter.php‘);
3.  
   class Publisher extends Parenter
4.  
5.  
   public function __construct()
6.  
7.  
   parent::__construct(‘exchange‘, ‘‘, ‘routeKey‘);
8.  
9.  
   public function doProcess($msg)
10.  
11.  
     
12.  
13.  
     
14.  

test.php（测试数据）

1.  
   <?php
2.  
   include_once(‘Publisher.php‘);
3.  
   $publisher = new Publisher();
4.  
   $publisher->sendMessage(‘Hello,World!‘);
5.  
   $publisher->closeConnetct();

5、添加交换机与队列

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

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

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

6、运行代码

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

运行测试脚本

如果打印出来字符串就成功了

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

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

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

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

            代码里很多注释都是我后来学习php-amqplib库中类的方法时加的，表示的是参数的意义，大家也可以去研究一下，这里提供个网址： Rabbitmq各方法的作用详解

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

下一篇：RabbitMQ的持久化（六）