快捷導(dǎo)航

Python隊(duì)列RabbitMQ 使用方法實(shí)例記錄

更新時(shí)間：2019年08月05日 10:39:21 作者：飛起的人生

這篇文章主要介紹了Python隊(duì)列RabbitMQ 使用方法,結(jié)合實(shí)例形式分析了Python隊(duì)列RabbitMQ創(chuàng)建隊(duì)列發(fā)送消息與創(chuàng)建消費(fèi)者消費(fèi)信息相關(guān)操作技巧,需要的朋友可以參考下

本文實(shí)例講述了Python隊(duì)列RabbitMQ 使用方法。分享給大家供大家參考，具體如下：

目前的exchange的路由策略是：每個(gè)需要隊(duì)列的服務(wù)獨(dú)享一個(gè)隊(duì)列（queue），消費(fèi)者（consumer）采用ACK自動(dòng)應(yīng)答模式處理隊(duì)列消息。

如果需要新增一個(gè)隊(duì)列服務(wù)，需要做如下開發(fā)步驟：

1.創(chuàng)建隊(duì)列，發(fā)送消息

<?php
$routingkey = 'key';
//設(shè)置你的連接
$conn_args = array('host' => 'localhost', 'port' => '5672', 'login' => 'guest', 'password' => 'guest');
$conn = new AMQPConnection($conn_args);
if ($conn->connect()) {
  echo "Established a connection to the broker \n";
} else {
  echo "Cannot connect to the broker \n ";
}
//你的消息
$message = json_encode(array('Hello World3!', 'php3', 'c++3:'));
//創(chuàng)建channel
$channel = new AMQPChannel($conn);
//創(chuàng)建exchange
$ex = new AMQPExchange($channel);
$ex->setName('exchange2'); //創(chuàng)建名字
$ex->setType(AMQP_EX_TYPE_DIRECT);
$ex->setFlags(AMQP_DURABLE);
echo "exchange2 status:" . $ex->declareExchange();
echo "\n";
for ($i = 0; $i < 100; $i++) {
  if ($routingkey == 'key2') {
    $routingkey = 'key';
  } else {
    $routingkey = 'key2';
  }
  $ex->publish($message, $routingkey);
}

這樣就產(chǎn)生了50條消息，但是沒有消費(fèi)者，所以沒有被消費(fèi)

2.創(chuàng)建消費(fèi)者，消費(fèi)信息

<?php
set_time_limit(0);
$e_name = 'exchange2'; //交換機(jī)名
$q_name = 'queue2'; //隊(duì)列名
$k_route = 'key2'; //路由key 
//連接RabbitMQ
$conn_args = array('host' => '127.0.0.1', 'port' => '5672', 'login' => 'guest', 'password' => 'guest', 'vhost' => '/');
$conn = new AMQPConnection($conn_args);
$conn->connect();
$channel = new AMQPChannel($conn);  
//創(chuàng)建交換機(jī)
$ex = new AMQPExchange($channel);
$ex->setName($e_name);
$ex->setType(AMQP_EX_TYPE_DIRECT); //direct類型
$ex->setFlags(AMQP_DURABLE); //持久化
echo "Exchange Status:" . $ex->declareExchange() . "\n";
//創(chuàng)建隊(duì)列
$q = new AMQPQueue($channel);
$q->setName($q_name);
$q->setFlags(AMQP_DURABLE); //持久化 
//綁定交換機(jī)與隊(duì)列，并指定路由鍵
echo 'Queue Bind: ' . $q->bind($e_name, $k_route) . "\n"; //阻塞模式接收消息
echo "Message:\n";
$q->consume('processMessage', AMQP_AUTOACK); //自動(dòng)ACK應(yīng)答 
$conn->disconnect();
/** * 消費(fèi)回調(diào)函數(shù) * 處理消息 */
function processMessage($envelope, $queue) {
  var_dump($envelope->getRoutingKey());
  $msg = $envelope->getBody();
  echo $msg . "\n"; //處理消息
}

運(yùn)行之后