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

网络编程 2024/9/25 佚名

3 2 1

无为清净楼资源网 Design By www.qnjia.com

前言

为什么使用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

 <"htmlcode">

 <"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




 <"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




 <"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()) ;
 }


总结
以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作具有一定的参考学习价值，谢谢大家对的支持。