3

我有一个有效的 PHP 脚本，它向 RabbitMQ 发布消息，它从制表符分隔的文本文件中解析出来。我从字面上将工作代码从该文件复制/粘贴到另一个文件，并希望建立一个消费者来检索发布到交换的消息，将它们 json_decode 并将它们插入数据库。

每次尝试，甚至从 PHP.net 站点复制/粘贴示例代码，甚至 SO 中的示例，都会失败并显示空白屏幕且没有错误消息，然后它甚至会杀死 php-fpm 进程。

知道为什么队列不会绑定以及这里出了什么问题吗？

  • Nginx -> php-fpm
  • PHP 5.3.x
  • Macbook Pro (OSX Lion)
  • RabbitMQ（已安装 librabbitmq 和 pecl amqp）

这是我尝试过的一个示例，但我在 AMQP 文档上尝试了 PHP.net 和 SO 示例，但没有任何工作。我可以很好地发布，但是当我尝试绑定队列时它会失败并最终 php-fpm 锁定。

<?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();
?>
4

1 回答 1

0

您是否尝试过此处找到的示例：https ://github.com/rabbitmq/rabbitmq-tutorials/tree/master/php ？

您要查看的两个是：emit_log.php receive_logs.php

使用的库与 PECL/内置库不同，我相信内置库不支持消费。

于 2012-07-29T16:04:03.117 回答