PHP+RabbitMQ实现消息队列的完整代码

日期：2020-05-21 栏目：程序人生浏览：次

前言

为什么使用RabbitMq而不是ActiveMq或者RocketMq？

首先，从业务上来讲，我并不要求消息的100%接受率，并且，我需要结合php开发，RabbitMq相较RocketMq，延迟较低(微妙级)。至于ActiveMq，貌似问题较多。RabbitMq对各种语言的支持较好，所以选择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;
  }
 }

转载注明出处：http://www.heiqu.com/5707.html

PHP+RabbitMQ实现消息队列的完整代码

相关推荐