本篇文章给大家带来的内容是关于php和rabbitmq实现消息队列的完整代码，有一定的参考价值，有需要的朋友可以参考一下，希望对你有所帮助。
先安装php对应的rabbitmq,这里用的是 php_amqp 不同的扩展实现方式会有细微的差异.
php扩展地址: http://pecl.php.net/package/amqp
具体以官网为准  http://www.rabbitmq.com/getstarted.html 
介绍
config.php 配置信息
basemq.php mq基类
productmq.php 生产者类
consumermq.php 消费者类
consumer2mq.php 消费者2(可有多个)
config.php    <?php return [ //配置 'host' => [            '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 'config.php' ; if(!$conf) throw new \amqpconnectionexception('config error!'); $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 'basemq.php'; class productmq extends basemq { private $routes = ['hello','word']; //路由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 'basemq.php'; class consumermq extends basemq { private $q_name = 'hello'; //队列名 private $route = 'hello'; //路由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()) ;    }
以上就是php和rabbitmq实现消息队列的完整代码的详细内容。