标签: php-amqplib

我已经设置了RabbitMQ,以便从外部API解析大约20.000个请求,但它会在几分钟后保持超时.它确实可以正确解析总共20.000个请求中的2000个.

日志文件说:

=INFO REPORT==== 16-Feb-2016::17:02:50 ===
accepting AMQP connection <0.1648.0> (127.0.0.1:33091 -> 127.0.0.1:5672)

=ERROR REPORT==== 16-Feb-2016::17:03:21 ===
closing AMQP connection <0.1648.0> (127.0.0.1:33091 -> 127.0.0.1:5672):
{writer,send_failed,{error,timeout}}

我已经增加了心跳值,但我无法弄清楚为什么它超时.配置为:Ubuntu 14.04,NGINX 1.8.1,RabbitMQ 3.6.0

我很感激你的时间和投入!

我需要创建一个简单的队列管理器来将数字从发件人传递给消费者.RabbitMQ提供的Hello World教程几乎覆盖了70%.

但我需要将队列更改为永远不会等待传入的消息.或者在一定数量的消息后停止等待.我阅读并尝试了其他帖子中的一些解决方案,但它不起作用.

rabbitmq AMQP :: consume() - 未定义的方法.还有另一种方法,wait_frame但它受到保护.

和其他帖子是python,我不明白.

<?php

require_once __DIR__ . '/vendor/autoload.php';
require 'config.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

function recieveQueue($queueName){
    $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');

    // try{
    //  $connection->wait_frame(10);
    // }catch(AMQPConnectionException $e){
    //  echo "asdasd";
    // }

    $channel = $connection->channel();

    $channel->queue_declare($queueName, false, false, false, false);

    echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";

    $callback = function($msg) {
        echo " [x] Received ", $msg->body, "\n";

    };

    // $tag = uniqid() . microtime(true);
    // $queue->consume($callback, $flags, …

I'm trying to use the Delayed Message Queue for RabbitMQ from PHP, but my messages are simply disappearing.

I'm declaring the exchange with the following code:

$this->channel->exchange_declare(
    'delay',
    'x-delayed-message',
    false,  /* passive, create if exchange doesn't exist */
    true,   /* durable, persist through server reboots */
    false,  /* autodelete */
    false,  /* internal */
    false,  /* nowait */
    ['x-delayed-type' => ['S', 'direct']]);

I'm binding the queue with this code:

$this->channel->queue_declare(
    $queueName,
    false,  /* Passive */
    true,   /* Durable */
    false, …

我需要从一个队列中捕获一条实际消息.兔子试图捕捉所有这些.简化代码如下:

private function getSingleTask(){
$connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');

$channel = $connection->channel();
$channel->queue_declare('hello', false, false, false, false);

$callback = function($msg) {
 return $msg->body;
};

$channel->basic_qos(null, 1, null);
$channel->basic_consume('helloQueue', '', false, true, false, false, $callback);
$channel->wait(null, true, 5);
}

我向队列扔了几条消息,但是一旦我执行下面的部分代码,它就会从队列和$回调中获取所有消息.

我有一个简单的排队系统,显然,它接收消息并发布它们.

但是,由于系统中的新开发,我们现在需要检查x-death来自交换的头文件,但是,我似乎无法找到有关如何通过PHP AMQP库检索它的任何文档.

任何人对如何实现这一点有任何想法？

当尝试从 RabbitMQ 接收答案时，我收到错误。

2015-07-03 09:31:12 [10.1.101.55][-][-][error][PhpAmqpLib\Exception\AMQPTimeoutException] exception 'PhpAmqpLib\Exception\AMQPTimeoutException' with message 'The connection timed out after 30 sec while awaiting incoming data' in /opt/www/site.ll/vendor/php-amqplib/PhpAmqpLib/Wire/AMQPReader.php:130
Stack trace:
#0 /opt/www/site.ll/vendor/php-amqplib/PhpAmqpLib/Wire/AMQPReader.php(147): PhpAmqpLib\Wire\AMQPReader->wait()
#1 /opt/www/site.ll/vendor/php-amqplib/PhpAmqpLib/Wire/AMQPReader.php(105): PhpAmqpLib\Wire\AMQPReader->rawread(7)
#2 /opt/www/site.ll/vendor/php-amqplib/PhpAmqpLib/Connection/AbstractConnection.php(528): PhpAmqpLib\Wire\AMQPReader->read(7)
#3 /opt/www/site.ll/vendor/php-amqplib/PhpAmqpLib/Connection/AbstractConnection.php(568): PhpAmqpLib\Connection\AbstractConnection->wait_frame(30)
#4 /opt/www/site.ll/vendor/php-amqplib/PhpAmqpLib/Channel/AbstractChannel.php(198): PhpAmqpLib\Connection\AbstractConnection->wait_channel(1, 30)
#5 /opt/www/site.ll/vendor/php-amqplib/PhpAmqpLib/Channel/AbstractChannel.php(332): PhpAmqpLib\Channel\AbstractChannel->next_frame(30)

当我尝试从本地电脑接收答案时，它可以工作，但在生产服务器上 - 不行。

为什么，以及我该如何解决它。

谢谢！

我在RabbitMQ上出现了染色错误。我发送了一些关于RabbitMq和Php的产品信息，但是出了点问题。我可以在异常“ errno = 104对等重置连接”之后的一分钟后发送消息。见下文：

2016-05-09 08:13:14 Daemon start
2016-05-09 08:13:17 Send to shop-id-1-price
2016-05-09 08:13:21 Send to shop-id-1-price
2016-05-09 08:13:24 Send to shop-id-1-price
2016-05-09 08:13:28 Send to shop-id-1-price
2016-05-09 08:13:32 Send to shop-id-1-price
2016-05-09 08:13:36 Send to shop-id-1-price
2016-05-09 08:13:39 Send to shop-id-1-price
2016-05-09 08:13:43 Send to shop-id-1-price
2016-05-09 08:13:47 Send to shop-id-1-price


  [ErrorException]                                                           
  fwrite(): send of 21 bytes failed with errno=104 Connection reset by peer  


rabbitmq:consumer [-m|--messages [MESSAGES]] [-r|--route [ROUTE]] [-l|--memory-limi
t [MEMORY-LIMIT]] [-d|--debug] [--] <name>

PHP Fatal error: …

我对 RabbitMQ 还很陌生。我正在使用 codeigniter 来使用 php-amqplib 库，并且仍然想知道我所缺乏的一些知识。

• 为什么$channel->wait()使用？
• 为什么它总是驻留在无限的 while 循环中？
• 如何/我可以绕过无限 while 循环。

就像在我的项目的一个用户想要向 100k 潜在客户广播新活动的情况下，如果第二个用户要发送大约 100 封邮件，那么第二个用户就会受到影响，第二个用户必须先等待 100k 邮件发送，然后是最后一个用户轮到他了。

我需要一个针对并发消费者的解决方案，使他们能够顺利工作而不影响其他人

这是我的代码片段：

public function campaign2(){
        $this->load->library('mylibrary');
        for( $i=1;$i<=5;$i++ ) {
            $url = "http://localhost/myproject/rabbit/waiting";
            $param = array('index' => $i);
            $this->waiting($i);
        }          
}

public function waiting($i)
    {
        ini_set('memory_limit','400M');
        ini_set('max_execution_time', 0);
        ini_set('display_errors', 1);

        ${'conn_'.$i} = connectRabbit();
        ${'channel_'.$i} = ${'conn_'.$i}->channel();
        ${'channel_'.$i}->exchange_declare('ha-local-campaign-'.$i.'-exchange', 'fanout', false, true, false);
        $q    = populateQueueName('campaign-'.$i);
        ${'channel_'.$i}->queue_declare($q, false, true, false, false); 
        ${'channel_'.$i}->queue_bind($q, 'ha-local-campaign-'.$i.'-exchange', 'priority.'.$i);
        $consumer_tag = 'campaign_consumer' ;
        function …

添加 'delivery_mode' => 2 后开始出现错误不可恢复的错误：PreconditionFailed(406, 'PRECONDITION_FAILED - 消费者在通道 1 上确认超时', (0, 0), '') 使用的超时值：1800000 毫秒。这个超时值可以配置，请参阅消费者文档指南以了解更多信息

它间歇性地抛出此错误并且 Pod 正在重新启动。

我正在使用RabbitMQ和Node.JS开发一个消息传递应用程序.我为此目的使用amqplib.我是Node.JS的新手,并且在理解amqplib的语法方面遇到了一些困难.例如,有一个声明队列的函数,即

channel.assertQueue([queue, [options, [function(err, ok) {...}]]]);

我在过去的2-3天里一直在提这个,但我仍然不清楚这些 - > err和ok.如何使用这些参数？

一个例子将非常感激.