快捷導(dǎo)航

Python隊列RabbitMQ 使用方法實例記錄

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

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

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

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

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

1.創(chuàng)建隊列，發(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條消息，但是沒有消費者，所以沒有被消費

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

<?php
set_time_limit(0);
$e_name = 'exchange2'; //交換機名
$q_name = 'queue2'; //隊列名
$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)建交換機
$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)建隊列
$q = new AMQPQueue($channel);
$q->setName($q_name);
$q->setFlags(AMQP_DURABLE); //持久化 
//綁定交換機與隊列，并指定路由鍵
echo 'Queue Bind: ' . $q->bind($e_name, $k_route) . "\n"; //阻塞模式接收消息
echo "Message:\n";
$q->consume('processMessage', AMQP_AUTOACK); //自動ACK應(yīng)答 
$conn->disconnect();
/** * 消費回調(diào)函數(shù) * 處理消息 */
function processMessage($envelope, $queue) {
  var_dump($envelope->getRoutingKey());
  $msg = $envelope->getBody();
  echo $msg . "\n"; //處理消息
}

運行之后