PHP中RabbitMQ之phpAmqplib實現（五

本章講訴如何使用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的持久化（六）