?

?前言：?

?

? ? 之前有寫過死信隊(duì)列的使用場景以及通過管控臺(tái)創(chuàng)建死信。這次就通過代碼實(shí)現(xiàn)死信隊(duì)列的創(chuàng)建，同時(shí)也分享一下RabbitMQ封裝的類。

準(zhǔn)備：

1. 先準(zhǔn)備一個(gè)死信隊(duì)列(最后用來消費(fèi))的參數(shù)配置，包括虛擬機(jī)，交換機(jī)，隊(duì)列，有效時(shí)間等，如下。

2. 按照上面在RabbitMQ中創(chuàng)建虛擬機(jī)和交換機(jī)，死信隊(duì)列。并讓交換機(jī)與死信隊(duì)列綁定，操作方法前面有介紹。

3. 這里就直接提供rabbitMQ操作的基本封裝的類，包括一個(gè)基類，生產(chǎn)者類，消費(fèi)者類。

3.1. 基類。

<?php namespace rabbitmq;class BaseMQ{public $AMQPChannel ;public $AMQPConnection ;public $AMQPEnvelope ;public $AMQPExchange ;public $AMQPQueue ;public $conf ;public $exchange ;public $queue;public $route;public $queueArgs;public function __construct($host,$options,$args = []){
$config = include 'config/config.php';if (!$config)throw new \AMQPConnectionException('config error!');$this->host = array_merge($config,$host);isset($options['vhost']) && $this->host['vhost'] = $options['vhost'];$this->exchange = $options['exchange'];$this->queue = $options['queue'];$this->route = $options['route'];$this->queueArgs = $args;$this->AMQPConnection = new \AMQPConnection($this->host);if (!$this->AMQPConnection->connect())throw new \AMQPConnectionException("Cannot connect to the broker!\n");
}public function close(){$this->AMQPConnection->disconnect();
}public function channel(){if (!$this->AMQPChannel) {$this->AMQPChannel = new \AMQPChannel($this->AMQPConnection);
}return $this->AMQPChannel;
}public function exchange(){if (!$this->AMQPExchange) {$this->AMQPExchange = new \AMQPExchange($this->channel());$this->AMQPExchange->setName($this->exchange);
}return $this->AMQPExchange ;
}public function queue(){if (!$this->AMQPQueue) {$this->AMQPQueue = new \AMQPQueue($this->channel());
}return $this->AMQPQueue ;
}public function envelope(){if (!$this->AMQPEnvelope) {$this->AMQPEnvelope = new \AMQPEnvelope();
}return $this->AMQPEnvelope;
}
}

3.2. 生產(chǎn)者類。

<?php namespace rabbitmq;class ProductMQ extends BaseMQ{/** 只控制發(fā)送成功 不接受消費(fèi)者是否收到
* @throws \AMQPChannelException
* @throws \AMQPConnectionException
* @throws \AMQPExchangeException
*/public function publish($message){
$message = is_array($message)?json_encode($message):$message;//頻道
$channel = $this->channel();//創(chuàng)建交換機(jī)對象
$ex = $this->exchange();return $ex->publish($message, $this->route, AMQP_NOPARAM, array('delivery_mode' => 2));
}
}

3.3. 消費(fèi)者。

<?php namespace rabbitmq;class ConsumerMQ extends BaseMQ{public function run($processMessage){// 創(chuàng)建交換機(jī)
$ex = $this->exchange();// direct類型
$ex->setType(AMQP_EX_TYPE_DIRECT); // 持久化
$ex->setFlags(AMQP_DURABLE);// 不存在就發(fā)布
$ex->declareExchange();// 創(chuàng)建隊(duì)列
$q = $this->queue();// 設(shè)置隊(duì)列名稱
$q->setName($this->queue);// 持久化
$q->setFlags(AMQP_DURABLE); // 隊(duì)列參數(shù)
is_array($this->queueArgs) && $q->setArguments($this->queueArgs);
$q->declareQueue();//綁定交換機(jī)與隊(duì)列，并指定路由鍵// echo 'Queue Bind: '.$q->bind($this->exchange, $this->route)."\n";
$q->bind($this->exchange, $this->route);//阻塞模式接收消息// echo "Message:\n";if (!is_null($processMessage)) {while (True) {
$q->consume($processMessage);
}
}$this->close();
}
}編碼：

? ? ?上面的死信隊(duì)列已經(jīng)創(chuàng)建好了，接下來主要就是通過代碼創(chuàng)建一個(gè)用于直接生產(chǎn)消息的普通隊(duì)列，但是這個(gè)隊(duì)列需要設(shè)置三個(gè)參數(shù)。

x-dead-letter-exchange：關(guān)聯(lián)死信的交換機(jī)x-dead-letter-routing-key 關(guān)聯(lián)死信的路由keyx-message-ttl 當(dāng)前隊(duì)列消息的有效期，也就是多久后消息自動(dòng)進(jìn)行死信隊(duì)列，并且從本隊(duì)列刪除

? ?1. 代碼部分：

public function addToDlx()
{
$host = [
'host' => '127.0.0.1',
'port' => '5672',
'login' => 'guest',
'password' => 'guest',
'vhost' => 'report',
'heartbeat' => 60
];

// 普通隊(duì)列
$normal = [
'vhost' => 'report', // 虛擬機(jī)
'exchange' => 'normal', // 交換機(jī)
'route' => 'normal_route', // 路由key - 用于交換機(jī)與隊(duì)列進(jìn)行綁定
'queue' => 'normal_queue', // 隊(duì)列
'expire' => 1000*60, // 有效時(shí)間單位：毫秒 - 1分鐘
];

// 死信隊(duì)列
$normal_dlx = [
'vhost' => 'report',
'exchange' => 'normal_dlx',
'route' => 'normal_dlx_route',
'queue' => 'normal_dlx_queue'
];

// 給普通隊(duì)列關(guān)聯(lián)死信隊(duì)列，攜帶的參數(shù)
$dlx_args = [
'x-dead-letter-exchange' => $normal_dlx['exchange'],
'x-dead-letter-routing-key' => $normal_dlx['route'],
'x-message-ttl' => $normal['expire'],
];

通過消費(fèi)者方式創(chuàng)建死信隊(duì)列/
$dlx_mq = new ConsumerMQ($host,$normal,$dlx_args);
$dlx_mq->run(null);

//// 將消息放入普通隊(duì)列/
$mq = new ProductMQ($host, $normal);

$param = json_encode([
'name' => 'test',
'id' => 11568,
'remark' => '測試一下'
]);

$mq->publish($param);

$mq->close();
}

2. 測試結(jié)果：

? ? ? ?通過postman點(diǎn)擊上面接口，控制臺(tái)就可以看出多出了一個(gè)normal隊(duì)列，并且隊(duì)列的?Features?為“ D?TTL?DLX?DLK ”，$param的消息也會(huì)首先進(jìn)入“normal”隊(duì)列。

2. 一分鐘后(自己設(shè)置的)，normal的消息會(huì)失效，進(jìn)而開始添加到了死信隊(duì)列“normal_dxl”，可以點(diǎn)擊死信查看最新的消息信息。

總結(jié)

以上是生活随笔為你收集整理的消息队列控制灯代码_代码实现RabbitMQ死信队列的创建的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。