我有一个可以发布信息到RabbitMQ的PHP脚本,这些信息是从一个制表符分隔的文本文件中解析出来的。我直接将这个工作正常的代码从这个文件复制/粘贴到另一个文件,并希望建立一个消费者来获取发布到交换机的消息,对它们进行json_decode并将它们插入到数据库中。
所有尝试,甚至从PHP.net网站和SO中复制/粘贴示例代码都失败了,仅显示一个空白屏幕而没有错误消息,这甚至会导致php-fpm进程关闭。
请问为什么队列不会绑定,这里出了什么问题?
以下是一种尝试方法的示例,但我已经尝试过AMQP文档上PHP.net和SO的示例,但都无法工作。我可以正常发布,但当我尝试绑定队列时,它失败并且最终php-fpm被锁住了。
所有尝试,甚至从PHP.net网站和SO中复制/粘贴示例代码都失败了,仅显示一个空白屏幕而没有错误消息,这甚至会导致php-fpm进程关闭。
请问为什么队列不会绑定,这里出了什么问题?
以下是一种尝试方法的示例,但我已经尝试过AMQP文档上PHP.net和SO的示例,但都无法工作。我可以正常发布,但当我尝试绑定队列时,它失败并且最终php-fpm被锁住了。
- Nginx -> php-fpm
- PHP 5.3.x
- Macbook Pro (OSX Lion)
- RabbitMQ (已安装librabbitmq和pecl amqp)
<?php
// Report all PHP errors
error_reporting(E_ALL);
/*****************************************
* MQ settings
****************************************/
$mq = array(
'host' => 'localhost',
'port' => 5672,
'login' => 'guest',
'password' => 'guest',
'exchange' => 'gbus.user',
'routing_key' => 'gbus.test.mike',
);
/*****************************************
* Connect to queue
****************************************/
$conn_args = array('host' => $mq['host'], 'port' => $mq['port'], 'login' => $mq['login'], 'password' => $mq['password']);
$conn = new AMQPConnection($conn_args);
$conn->connect();
$ch = new AMQPChannel($conn);
// Create a new queue
$q = new AMQPQueue($ch);
$q->declare('test-queue');
$q->bind($mq['exchange'],$mq['routing_key']);
?>
<br>
<font color="blue" face="arial" size="4">File Contents</font>
<hr>
<?php
while(true){
$msg=$q->get();
if ($msg['count']>-1){
echo "\n--------\n";
print_r($msg['msg']);
echo "\n--------\n";
}
sleep(1);
}
if (!$conn->disconnect()) {
throw new Exception('Could not disconnect');
}
?>
这是我曾经用来发布到队列的示例代码,每次运行此代码后,我可以在RabbitMQ控制面板中看到20条新消息。出于测试目的,我将其限制为20条,但文件有成千上万行。
可用的发布代码:
<?php
/*****************************************
* MQ settings
****************************************/
$mq = array(
'host' => 'localhost',
'port' => 5672,
'login' => 'guest',
'password' => 'guest',
'exchange' => 'gbus.user',
'routing_key' => 'gbus.test.mike',
);
/*****************************************
* Connect to queue
****************************************/
$conn_args = array('host' => $mq['host'], 'port' => $mq['port'], 'login' => $mq['login'], 'password' => $mq['password']);
$conn = new AMQPConnection($conn_args);
$conn->connect();
$ch = new AMQPChannel($conn);
$ex = new AMQPExchange($ch);
$ex->setName($mq['exchange']);
/*****************************************
* Parse the file
****************************************/
$filename = "/tmp/Users.txt";
$board = "test";
$fd = fopen ($filename, "r");
$contents = fread ($fd,filesize ($filename));
fclose ($fd);
$delimiter = "\r\n";
$rows = explode($delimiter, $contents);
$counter = 0;
?>
<br>
<font color="blue" face="arial" size="4">File Rows (first 20)</font>
<hr>
<?php
foreach ( $rows as $row )
{
$counter++;
echo "<b>Row $counter: </b> $row<br>";
// build list columns
list($login_name, $pwd, $account_type, $access_level, $status, $first_name, $last_name, $agent_code) = explode("\t", $row);
// build assoc array for json
$user = array("domain"=>$board, "username"=>$login_name, "user_id"=>$agent_code, "password"=>$pwd, "first_name"=>$first_name, "last_name"=>$last_name);
// Publish a message to the exchange with a routing key
$ex->publish(json_encode($user), $mq['routing_key'], AMQP_NOPARAM, array("content_type"=>"application/data"));
if($counter == 20) {
break;
}
}
$ch->close();
$conn->close();
?>
rabbitmqctl list_queues
列出你的队列,使用list_exchanges列出交换机,使用list_bindings获取调试信息以帮助你解决问题。请查看http://www.rabbitmq.com/man/rabbitmqctl.1.man.html获取更多详细信息。将转储文件发布在此处,这将有助于找到问题所在。 - MaX