快速使用PHP的RabbitMq實例

1.安裝

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

 

composer require php-amqplib/php-amqplib

2. 使用

新建 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 消費

<?php
require "RabbitMq.php";

class Consumer extends RabbitMqService
{
    public function __construct()
    {
        parent::__construct('exchange', 'queue', 'routeKey');
    }

    public function doProcess($msg)
    {
        echo $msg . "\n";
    }
}

$consumer = new Consumer();
//$consumer->dealMq(false);
$consumer->dealMq(false);

　　

3. 創建生產者 Publisher .php

<?php
require "RabbitMq.php";

class Publisher extends RabbitMqService
{
    public function __construct()
    {
        parent::__construct('exchange', '', 'routeKey');
    }

    public function doProcess($msg)
    {

    }

}

$publisher = new Publisher();
$publisher->sendMessage('Hello,World!');

$publisher->closeConnetct();

　　

完成之后進行調試

1.先啟動消費者

 

2.在啟動生產者發送數據

 

 

 3.查看消費者這邊

 

 

也可以到 http://127.0.0.1:15672/ 本機RabbitMq 管理查看連接情況，隊列數據等