1. 首页
  2. 每日优选

为什么使用RabbitMq而不是ActiveMq或者RocketMq实现消息队列

为什么使用RabbitMq而不是ActiveMq或者RocketMq实现消息队列

文章来自:脚本之家
链接:https://www.jb51.net/article/158162.htm

作者:磊丰

商务合作:请加微信(QQ):2230304070










































为什么使用RabbitMq而不是ActiveMq或者RocketMq实现消息队列

资源教程分享

为什么使用RabbitMq而不是ActiveMq或者RocketMq实现消息队列

码农网-技术教程资源分享平台:http://www.mano100.cn。这是一个资源教程分享网,网站收集了php视频教程,前端各种视频教程,接口API,微信公众号,支付宝视频教程,还有go语言,python等视频教程,教程以链接+提取码的方式分享与大家!

用公众号每天来阅读文章,学习,目前是最方便的。这是自闲暇的时间里,利用5分钟读读技术文章,丰富自己的知识。。。当然,学习有很多种方法,不单单是基于公众号来学习,你可以到网上查找更多的教程,学习文档,技术书本,视频教程!各有各的爱好吧!

视频教程推荐

RabbitMQ消息中间件视频教程

链接:http://www.mano100.cn/thread-228-1-1.html

实战消息队列中间件视频教程

链接:http://www.mano100.cn/thread-229-1-1.html

RabbitMQ消息中间件技术精讲

链接:http://www.mano100.cn/thread-246-1-1.html

金币充值规则与优惠

链接:http://www.mano100.cn/thread-169-1-1.html







为什么使用RabbitMq而不是ActiveMq或者RocketMq实现消息队列

文章正文

为什么使用RabbitMq而不是ActiveMq或者RocketMq实现消息队列

为什么使用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: 2019/07/13
 * Time: 14:11
 */


namespace MyObjSummaryrabbitMQ;

/** Member
 *  AMQPChannel
 *  AMQPConnection
 *  AMQPEnvelope
 *  AMQPExchange
 *  AMQPQueue
 * Class BaseMQ
 * @package MyObjSummaryrabbitMQ
 */

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

以上是文章全部内容,有需要学习与经验交流的友友可以加小编。咱们一起学习,有问题一起交流,一起进步!


让学习成为一种习惯

长按二维码加好友 

为什么使用RabbitMq而不是ActiveMq或者RocketMq实现消息队列

学习  |  交流  |  分享  |  文章

原创文章,作者:比特币区块链日报,如若转载,请注明出处:https://www.dailybtc.cn/%e4%b8%ba%e4%bb%80%e4%b9%88%e4%bd%bf%e7%94%a8rabbitmq%e8%80%8c%e4%b8%8d%e6%98%afactivemq%e6%88%96%e8%80%85rocketmq%e5%ae%9e%e7%8e%b0%e6%b6%88%e6%81%af%e9%98%9f%e5%88%97/

发表评论

电子邮件地址不会被公开。 必填项已用*标注

联系我们

在线咨询:点击这里给我发消息

邮件:[email protected]

工作时间:周一至周五,9:30-18:30,节假日休息

QR code