PHP+RabbitMQ实现消息队列的完整代码
前言
为什么使用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.phpMQ基类
- ProductMQ.php生产者类
- ConsumerMQ.php消费者类
- Consumer2MQ.php消费者2(可有多个)
config.php
[ 'host'=>'127.0.0.1', 'port'=>'5672', 'login'=>'guest', 'password'=>'guest', 'vhost'=>'/', ], //交换机 'exchange'=>'word', //路由 'routes'=>[], ];
BaseMQ.php
conf=$conf['host'];
$this->exchange=$conf['exchange'];
$this->AMQPConnection=new\AMQPConnection($this->conf);
if(!$this->AMQPConnection->connect())
thrownew\AMQPConnectionException("Cannotconnecttothebroker!\n");
}
/**
*closelink
*/
publicfunctionclose()
{
$this->AMQPConnection->disconnect();
}
/**Channel
*@return\AMQPChannel
*@throws\AMQPConnectionException
*/
publicfunctionchannel()
{
if(!$this->AMQPChannel){
$this->AMQPChannel=new\AMQPChannel($this->AMQPConnection);
}
return$this->AMQPChannel;
}
/**Exchange
*@return\AMQPExchange
*@throws\AMQPConnectionException
*@throws\AMQPExchangeException
*/
publicfunctionexchange()
{
if(!$this->AMQPExchange){
$this->AMQPExchange=new\AMQPExchange($this->channel());
$this->AMQPExchange->setName($this->exchange);
}
return$this->AMQPExchange;
}
/**queue
*@return\AMQPQueue
*@throws\AMQPConnectionException
*@throws\AMQPQueueException
*/
publicfunctionqueue()
{
if(!$this->AMQPQueue){
$this->AMQPQueue=new\AMQPQueue($this->channel());
}
return$this->AMQPQueue;
}
/**Envelope
*@return\AMQPEnvelope
*/
publicfunctionenvelope()
{
if(!$this->AMQPEnvelope){
$this->AMQPEnvelope=new\AMQPEnvelope();
}
return$this->AMQPEnvelope;
}
}
ProductMQ.php
channel();
//创建交换机对象
$ex=$this->exchange();
//消息内容
$message='productmessage'.rand(1,99999);
//开始事务
$channel->startTransaction();
$sendEd=true;
foreach($this->routesas$route){
$sendEd=$ex->publish($message,$route);
echo"SendMessage:".$sendEd."\n";
}
if(!$sendEd){
$channel->rollbackTransaction();
}
$channel->commitTransaction();//提交事务
$this->close();
die;
}
}
try{
(newProductMQ())->run();
}catch(\Exception$exception){
var_dump($exception->getMessage());
}
ConsumerMQ.php
exchange();
$ex->setType(AMQP_EX_TYPE_DIRECT);//direct类型
$ex->setFlags(AMQP_DURABLE);//持久化
//echo"ExchangeStatus:".$ex->declare()."\n";
//创建队列
$q=$this->queue();
//var_dump($q->declare());exit();
$q->setName($this->q_name);
$q->setFlags(AMQP_DURABLE);//持久化
//echo"MessageTotal:".$q->declareQueue()."\n";
//绑定交换机与队列，并指定路由键
echo'QueueBind:'.$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{
(newConsumerMQ)->run();
}catch(\Exception$exception){
var_dump($exception->getMessage());
}
总结
以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作具有一定的参考学习价值，谢谢大家对毛票票的支持。