PHP+RabbitMQ實(shí)現(xiàn)消息隊(duì)列的完整代碼

第PHP+RabbitMQ實(shí)現(xiàn)消息隊(duì)列的完整代碼為什么使用RabbitMq而不是ActiveMq或者RocketMq？

首先，從業(yè)務(wù)上來講，我并不要求消息的100%接受率，并且，我需要結(jié)合php開發(fā)，RabbitMq相較RocketMq，延遲較低(微妙級(jí))。至于ActiveMq，貌似問題較多。RabbitMq對(duì)各種語言的支持較好，所以選擇RabbitMq。

先安裝PHP對(duì)應(yīng)的RabbitMQ,這里用的是php_amqp不同的擴(kuò)展實(shí)現(xiàn)方式會(huì)有細(xì)微的差異.

php擴(kuò)展地址:/package/amqp

具體以官網(wǎng)為準(zhǔn)/getstarted.html

介紹

config.php配置信息

BaseMQ.phpMQ基類

ProductMQ.php生產(chǎn)者類

ConsumerMQ.php消費(fèi)者類

Consumer2MQ.php消費(fèi)者2(可有多個(gè))

config.php

return[

//配置

'host'=[

'host'='',

'port'='5672',

'login'='guest',

'password'='guest',

'vhost'='/',

//交換機(jī)

'exchange'='word',

//路由

'routes'=[],

];

BaseMQ.php

*CreatedbyPhpStorm.

*User:pc

*Date:2025/12/13

*Time:14:11

namespaceMyObjSummary\rabbitMQ;

/**Member

*AMQPChannel

*AMQPConnection

*AMQPEnvelope

*AMQPExchange

*AMQPQueue

*ClassBaseMQ

*@packageMyObjSummary\rabbitMQ

classBaseMQ

/**MQChannel

*@var\AMQPChannel

public$AMQPChannel;

/**MQLink

*@var\AMQPConnection

public$AMQPConnection;

/**MQEnvelope

*@var\AMQPEnvelope

public$AMQPEnvelope;

/**MQExchange

*@var\AMQPExchange

public$AMQPExchange;

/**MQQueue

*@var\AMQPQueue

public$AMQPQueue;

/**conf

*@var

public$conf;

/**exchange

*@var

public$exchange;

/**link

*BaseMQconstructor.

*@throws\AMQPConnectionException

publicfunction__construct()

$conf=require'config.php';

if(!$conf)

thrownew\AMQPConnectionException('configerror!');

$this-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

//生產(chǎn)者P

namespaceMyObjSummary\rabbitMQ;

require'BaseMQ.php';

classProductMQextendsBaseMQ

private$routes=['hello','word'];//路由key

*ProductMQconstructor.

*@throws\AMQPConnectionException

publicfunction__construct()

parent::__construct();

/**只控制發(fā)送成功不接受消費(fèi)者是否收到

*@throws\AMQPChannelException

*@throws\AMQPConnectionException

*@throws\AMQPExchangeException

publicfunctionrun()

//頻道

$channel=$this-channel();

//創(chuàng)建交換機(jī)對(duì)象

$ex=$this-exchange();

//消息內(nèi)容

$message='productmessage'.rand(1,99999);

//開始事務(wù)

$channel-startTransaction();

$sendEd=true;

foreach($this-routesas$route){

$sendEd=$ex-publish($message,$route);

echo"SendMessage:".$sendEd."\n";

if(!$sendEd){

$channel-rollbackTransaction();

$channel-commitTransaction();//提交事務(wù)

$this-close();

die;

try{

(newProductMQ())-run();

}catch(\Exception$exception){

var_dump($exception-getMessage());

}

ConsumerMQ.php

//消費(fèi)者C

namespaceMyObjSummary\rabbitMQ;

require'BaseMQ.php';

classConsumerMQextendsBaseMQ

private$q_name='hello';//隊(duì)列名

private$route='hello';//路由key

*ConsumerMQconstructor.

*@throws\AMQPConnectionException

publicfunction__construct()

parent::__construct();

/**接受消息如果終止重連時(shí)會(huì)有消息

*@throws\AMQPChannelException

*@throws\AMQPConnectionException

*@throws\AMQPExchangeException

*@throws\AMQPQueueException

publicfunctionrun()

//創(chuàng)建交換機(jī)

$ex=$this-exchange();

$ex-setType(AMQP_EX_TYPE_DIRECT);//direct類型

$ex-setFlags(AMQP_DURABLE);//持久化

//echo"ExchangeStatus:".$ex-declare()."\n";

//創(chuàng)建隊(duì)列

$q=$this-queue();

//var_dump($q-declare());exit();

$q-setName($this-q_name);

$q-setFlags(AMQP_DURABLE);//持久化

//echo"MessageTotal:".$q-declareQueue()."\n";

//綁定交換機(jī)與隊(duì)列，并指定路由鍵

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());//