Vollständiger Code zur Implementierung der Nachrichtenwarteschlange mit PHP und RabbitMQ

Was dieser Artikel Ihnen bringt, ist der vollständige Code für die Implementierung von Nachrichtenwarteschlangen in PHP und RabbitMQ. Ich hoffe, dass er Ihnen als Referenz dienen wird.

Installieren Sie zuerst RabbitMQ entsprechend PHP. Hier verwenden wir php_amqp. Verschiedene Erweiterungsimplementierungen weisen subtile Unterschiede auf.
PHP-Erweiterungsadresse: http://pecl.php.net/package/amqp
Für Einzelheiten finden Sie auf der offiziellen Website http://www.rabbitmq.com/getstarted.html

Einführung

config.php-Konfigurationsinformationen
BaseMQ.php MQ-Basisklasse
ProductMQ.php Producer-Klasse
ConsumerMQ.php Consumer-Klasse
Consumer2MQ.php Consumer 2 (kann mehrere haben)

config.php

    <?php
    return [
        //配置
        &#39;host&#39; => [
            'host' => '127.0.0.1',
            'port' => '5672',
            'login' => 'guest',
            'password' => 'guest',
            'vhost'=>'/',
        ],
        //交换机
        'exchange'=>'word',
        //路由
        'routes' => [],
    ];

BaseMQ.php

    <?php
    /**
     * Created by PhpStorm.
     * User: pc
     * Date: 2018/12/13
     * Time: 14:11
     */
    
    namespace MyObjSummary\rabbitMQ;
    
    /** Member
     *      AMQPChannel
     *      AMQPConnection
     *      AMQPEnvelope
     *      AMQPExchange
     *      AMQPQueue
     * Class BaseMQ
     * @package MyObjSummary\rabbitMQ
     */
    class BaseMQ
    {
        /** MQ Channel
         * @var \AMQPChannel
         */
        public $AMQPChannel ;
    
        /** MQ Link
         * @var \AMQPConnection
         */
        public $AMQPConnection ;
    
        /** MQ Envelope
         * @var \AMQPEnvelope
         */
        public $AMQPEnvelope ;
    
        /** MQ Exchange
         * @var \AMQPExchange
         */
        public $AMQPExchange ;
    
        /** MQ Queue
         * @var \AMQPQueue
         */
        public $AMQPQueue ;
    
        /** conf
         * @var
         */
        public $conf ;
    
        /** exchange
         * @var
         */
        public $exchange ;
    
        /** link
         * BaseMQ constructor.
         * @throws \AMQPConnectionException
         */
        public function __construct()
        {
            $conf =  require &#39;config.php&#39; ;
            if(!$conf)
                throw new \AMQPConnectionException(&#39;config error!&#39;);
            $this->conf     = $conf['host'] ;
            $this->exchange = $conf['exchange'] ;
            $this->AMQPConnection = new \AMQPConnection($this->conf);
            if (!$this->AMQPConnection->connect())
                throw new \AMQPConnectionException("Cannot connect to the broker!\n");
        }
    
        /**
         * close link
         */
        public function close()
        {
            $this->AMQPConnection->disconnect();
        }
    
        /** Channel
         * @return \AMQPChannel
         * @throws \AMQPConnectionException
         */
        public function channel()
        {
            if(!$this->AMQPChannel) {
                $this->AMQPChannel =  new \AMQPChannel($this->AMQPConnection);
            }
            return $this->AMQPChannel;
        }
    
        /** Exchange
         * @return \AMQPExchange
         * @throws \AMQPConnectionException
         * @throws \AMQPExchangeException
         */
        public function exchange()
        {
            if(!$this->AMQPExchange) {
                $this->AMQPExchange = new \AMQPExchange($this->channel());
                $this->AMQPExchange->setName($this->exchange);
            }
            return $this->AMQPExchange ;
        }
    
        /** queue
         * @return \AMQPQueue
         * @throws \AMQPConnectionException
         * @throws \AMQPQueueException
         */
        public function queue()
        {
            if(!$this->AMQPQueue) {
                $this->AMQPQueue = new \AMQPQueue($this->channel());
            }
            return $this->AMQPQueue ;
        }
    
        /** Envelope
         * @return \AMQPEnvelope
         */
        public function envelope()
        {
            if(!$this->AMQPEnvelope) {
                $this->AMQPEnvelope = new \AMQPEnvelope();
            }
            return $this->AMQPEnvelope;
        }
    }

ProductMQ.php

    <?php
    //生产者 P
    namespace MyObjSummary\rabbitMQ;
    require &#39;BaseMQ.php&#39;;
    class ProductMQ extends BaseMQ
    {
        private $routes = [&#39;hello&#39;,&#39;word&#39;]; //路由key
    
        /**
         * ProductMQ constructor.
         * @throws \AMQPConnectionException
         */
        public function __construct()
        {
           parent::__construct();
        }
    
        /** 只控制发送成功 不接受消费者是否收到
         * @throws \AMQPChannelException
         * @throws \AMQPConnectionException
         * @throws \AMQPExchangeException
         */
        public function run()
        {
            //频道
            $channel = $this->channel();
            //创建交换机对象
            $ex = $this->exchange();
            //消息内容
            $message = 'product message '.rand(1,99999);
            //开始事务
            $channel->startTransaction();
            $sendEd = true ;
            foreach ($this->routes as $route) {
                $sendEd = $ex->publish($message, $route) ;
                echo "Send Message:".$sendEd."\n";
            }
            if(!$sendEd) {
                $channel->rollbackTransaction();
            }
            $channel->commitTransaction(); //提交事务
            $this->close();
            die ;
        }
    }
    try{
        (new ProductMQ())->run();
    }catch (\Exception $exception){
        var_dump($exception->getMessage()) ;
    }

ConsumerMQ.php

    <?php
    //消费者 C
    namespace MyObjSummary\rabbitMQ;
    require &#39;BaseMQ.php&#39;;
    class ConsumerMQ extends BaseMQ
    {
        private  $q_name = &#39;hello&#39;; //队列名
        private  $route  = &#39;hello&#39;; //路由key
    
        /**
         * ConsumerMQ constructor.
         * @throws \AMQPConnectionException
         */
        public function __construct()
        {
            parent::__construct();
        }
    
        /** 接受消息 如果终止 重连时会有消息
         * @throws \AMQPChannelException
         * @throws \AMQPConnectionException
         * @throws \AMQPExchangeException
         * @throws \AMQPQueueException
         */
        public function run()
        {
    
            //创建交换机
            $ex = $this->exchange();
            $ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型
            $ex->setFlags(AMQP_DURABLE); //持久化
            //echo "Exchange Status:".$ex->declare()."\n";
    
            //创建队列
            $q = $this->queue();
            //var_dump($q->declare());exit();
            $q->setName($this->q_name);
            $q->setFlags(AMQP_DURABLE); //持久化
            //echo "Message Total:".$q->declareQueue()."\n";
    
            //绑定交换机与队列，并指定路由键
            echo 'Queue Bind: '.$q->bind($this->exchange, $this->route)."\n";
    
            //阻塞模式接收消息
            echo "Message:\n";
            while(True){
                $q->consume(function ($envelope,$queue){
                    $msg = $envelope->getBody();
                    echo $msg."\n"; //处理消息
                    $queue->ack($envelope->getDeliveryTag()); //手动发送ACK应答
                });
                //$q->consume('processMessage', AMQP_AUTOACK); //自动ACK应答
            }
            $this->close();
        }
    }
    try{
        (new ConsumerMQ)->run();
    }catch (\Exception $exception){
        var_dump($exception->getMessage()) ;
    }

Das obige ist der detaillierte Inhalt vonVollständiger Code zur Implementierung der Nachrichtenwarteschlange mit PHP und RabbitMQ. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!