Проблема с получением сообщений

[Vladimir Panivko]

День добрый.

Создал два файла один на отпарвку сообщейни второй на получение.

Вот файл на отправку
<?
$cnn = new AMQPConnect();
$exchange = new AMQPExchange($cnn);
$exchange->declare('HabraGuru', 'topic',AMQP_DURABLE );
$queue = new AMQPQueue($cnn);
$queue->declare('HabraGuru', AMQP_DURABLE);


$cnn = new AMQPConnect();
$exchange = new AMQPExchange($cnn, 'HabraGuru');
for ($i=0;$i<10;$i++){
    $msg = 'FriendLenta. How to. '.$i;
    $exchange->publish( $msg, 'presentation');
}
?>

Вот на получение
<?
$cnn = new AMQPConnect();
$exchange = new AMQPExchange($cnn);
$exchange->declare('HabraLuser', 'topic',AMQP_DURABLE );
$queue = new AMQPQueue($cnn, 'HabraLuser');
$queue->declare('HabraLuser', AMQP_DURABLE);
$queue->bind('HabraGuru', '*' );


$n = $queue->declare('HabraLuser', AMQP_DURABLE);
var_dump($n);
$res = $queue->consume($n);
var_dump($res);
for ($i=0;$i<=$n;$i++){
    $res = $queue->get();
    var_dump($res);
}

?>


Т.е. я отправляю 10 сообщений в очередь HabraGuru и расчитываю их получить по подписке в HabraLuser.
Вызов $n = $queue->declare('HabraLuser', AMQP_DURABLE); возвращает мне правильное количество отправленных сообщений, но при попытке их получить с помощью команды get я получаю false,
при попытке получить сообщения с помощью consume я получаю пустой массив.

В чём может быть проблема?

--
С уважением,
Владимир Панивко
+38 063 245-1651

[Alexandre Kalendarev]

 

<?
$cnn = new AMQPConnect();
$exchange = new AMQPExchange($cnn);

$exchange->declare('HabraGuru', 'topic',AMQP_DURABLE );

Создали обмен

 

$queue = new AMQPQueue($cnn);
$queue->declare('HabraGuru', AMQP_DURABLE);

создали очередь

$cnn = new AMQPConnect();
$exchange = new AMQPExchange($cnn, 'HabraGuru');

Эти две строчки выше  в контексте Вашей программы лишние

1) коннекция уже существует, зачем ее делать снова, используйте существующую

Вы же не коненктитесь к мускулю каждый раз при подаче второго и третьего запроса

2) Обмен уже объявлен - зачем его объявлять второй раз - это по аналогии с мускулем - зачем дважды делать CREATE TABLE

for ($i=0;$i<10;$i++){
    $msg = 'FriendLenta. How to. '.$i;
    $exchange->publish( $msg, 'presentation');
}

отправили 10 сообщений в обмен с ключом 'presentation'

Однако брокер (т.е. rabbitmq)  не знает в какую очередь из какого обмена и с каким ключом что переслать

Не хватает строчки типа 

$queue->Bind>('HabraGuru' , 'presentation');

 http://code.google.com/p/php-rabbit/wiki/AMQPQueueBind

 

Вот на получение
<?
$cnn = new AMQPConnect();
$exchange = new AMQPExchange($cnn);
$exchange->declare('HabraLuser', 'topic',AMQP_DURABLE );

Обмен уже объявлен - повторно его можно не объявлять

 

$queue = new AMQPQueue($cnn, 'HabraLuser');
$queue->declare('HabraLuser', AMQP_DURABLE);

очередь уже была создана и ее повторно можно не объявлять

 можно использовать обработку исключения на тот случай - если была объявлена очередь с флагом  AUTODELETE и она удалится когда будет пуста.

$queue->bind('HabraGuru', '*' );

 

А вот биндить надо до ОТПРАВКИ сообщения в обмен. Если биндим в даном случае, то к нам поступят все последующие сообщения после биндинга.

 

$n = $queue->declare('HabraLuser', AMQP_DURABLE);
var_dump($n);
$res = $queue->consume($n);
var_dump($res);
for ($i=0;$i<=$n;$i++){
    $res = $queue->get();
    var_dump($res);
}

 

Александр 

[Alexandre Kalendarev]

возможно была бага для определенных версий,

сейчас я ее устранил

в архиве рабочий клиент и экстеншен

если что -пиши

23 августа 2010 г. 2:47 пользователь Alexandre Kalendarev <aka...@gmail.com> написал:

[Владимир Панивко]

Спасибо проверю.

[Rauan Maemirov]

Будет ли работать данная схема?

$cnn = new AMQPConnect();

$exchange = new AMQPExchange($cnn);

$exchange->declare('exchange_1', 'topic', AMQP_DURABLE);

$exchange->bind('gurus', 'friends.#');

for ($i = 0; $i < 10; $i++) {

   $msg = 'FriendLenta. How to. ' . $i;

   $exchange->publish($msg, 'friends.feed');

}

Или создавать queue обязательно (хотя мы же пишем не в него, а в exchange)?

Нормально ли делать все заранее известные bind и declare в каком-нибудь отдельном скрипте? Чтобы код сводился к виду

$cnn = new AMQPConnect();

$exchange = new AMQPExchange($cnn, 'exchange_1');

for ($i = 0; $i < 10; $i++) {

    $msg = 'FriendLenta. How to. ' . $i;

    $exchange->publish($msg, 'friends.feed');

}

2010/9/20 Владимир Панивко <vpan...@gmail.com>

Спасибо проверю.

[Alexandre Kalendarev]

20 сентября 2010 г. 12:58 пользователь Rauan Maemirov <ra...@maemirov.com> написал:

Будет ли работать данная схема?

$cnn = new AMQPConnect();

$exchange = new AMQPExchange($cnn);

$exchange->declare('exchange_1', 'topic', AMQP_DURABLE);

$exchange->bind('gurus', 'friends.#');

 к этому моменту очередь должна быть создана.

 

for ($i = 0; $i < 10; $i++) {

   $msg = 'FriendLenta. How to. ' . $i;

   $exchange->publish($msg, 'friends.feed');

}

Или создавать queue обязательно (хотя мы же пишем не в него, а в exchange)?

Создавать queue обязательно, иначе может возникнуть ошибкак 404

это можно сделать только один раз!

 

Нормально ли делать все заранее известные bind и declare в каком-нибудь отдельном скрипте?

это нормально,

лишний раз объявление бинда, очереди или эксченджа - это лишний запрос к брокеру(раббиту)

можно сделать через эксепшен (если 404 ошибка, то делаем объявление и тому подобное )

[Rauan Maemirov]

Здравствуйте, Александр.

О классе Rabbit в доках не написано ни слова. Из примеров я понял, что это обертка над остальными AMQPConnect, AMQPExchange и так далее.

Как передать свои параметры для соединения?
Как передать deliveryTag для метода ack?

2010/9/19 Alexandre Kalendarev <aka...@gmail.com>

[Alexandre Kalendarev]

21 октября 2010 г. 11:41 пользователь Rauan Maemirov <ra...@maemirov.com> написал:

Здравствуйте, Александр.

О классе Rabbit в доках не написано ни слова. 

Первая версия был класс Rabbit, теперь семейство классов AMQPxxx

 

Из примеров я понял, что это обертка над остальными AMQPConnect, AMQPExchange и так далее.

Как передать свои параметры для соединения?

http://code.google.com/p/php-rabbit/wiki/AMQPConnect

$cnn = new AMQPConnect(array ('port'=>5673, 'login' => 'sector1' , 'host'=> '192.168.1.12', vhost='oylolo'));

Как передать deliveryTag для метода ack?

$e->publish( $msg, 'news' , 0,

array( 'Content-type'=>'text/json' , 

'Content-encoding'=>'UTF-8' ,

'user_id' => 'my_user123',

'message_id' => 'my_msg_123',

'delivery_mode' => 2,

'timestamp' => time(),

));

echo "len=".strlen($msg)."\n";

$msg = $q->get(0);

var_dump($msg);

// http://code.google.com/p/php-rabbit/wiki/AMQPQueueAck

$q->ack('1');

 

Alexandre

[Rauan Maemirov]

Какие из настроек должны быть указаны обязательно?

array( 'Content-type'=>'text/json' , 

'Content-encoding'=>'UTF-8' ,

'user_id' => 'my_user123',

'message_id' => 'my_msg_123',

'delivery_mode' => 2,

'timestamp' => time(),

)

Нужно ли указывать user_id, message_id, timestamp?

>$q->ack('1')

Что означает '1'? true?

2010/10/21 Alexandre Kalendarev <aka...@gmail.com>

[Rauan Maemirov]

Пытаюсь подключиться к кролику, выходит

PHP Fatal error:  Uncaught exception 'Exception' with message 'Socket error' in...

С rabbitmq 2.1.1 модуль не работает?

2010/9/19 Alexandre Kalendarev <aka...@gmail.com>

[Alexandre Kalendarev]

21 октября 2010 г. 13:17 пользователь Rauan Maemirov <ra...@maemirov.com> написал:

Какие из настроек должны быть указаны обязательно?

array( 'Content-type'=>'text/json' , 

'Content-encoding'=>'UTF-8' ,

'user_id' => 'my_user123',

'message_id' => 'my_msg_123',

'delivery_mode' => 2,

'timestamp' => time(),

)

 ни какие

все чисто информационные, можно их использовать для передачи доп информации подписчику

 

Нужно ли указывать user_id, message_id, timestamp?

нет, это по желанию,  timestamp отправления сообщения ни где не фиксируется, чисто для информации

>$q->ack('1')

Что означает '1'? true?

№ consumerTag (первое сообщение)

[Alexandre Kalendarev]

21 октября 2010 г. 15:42 пользователь Rauan Maemirov <ra...@maemirov.com> написал:

Пытаюсь подключиться к кролику, выходит

PHP Fatal error:  Uncaught exception 'Exception' with message 'Socket error' in...

проверь номер порта

 

С rabbitmq 2.1.1 модуль не работает?

нет не работал

модуль стабильно работает с версией до 2 (у меня 1.7), AMQP 0.8

все ни как не соберусь поставить 2-ю версию и погонять на ней.

[Alexandre Kalendarev]

>$q->ack('1')

Что означает '1'? true?

proto int queue::ack( long deliveryTag,[ bit params=AMQP_NONE ])

Возможный флаг  AMQP_MULTIPLE | AMQP_NONE default  

[Rauan Maemirov]

Сильно ли отличается 0.8 от 0.9.1?
Если использовать сейчас 1.7.2, не возникнет ли никаких проблем при переходе на 2.1?

>№ consumerTag (первое сообщение)
Что-то я все никак не врублюсь.
'1' - это сообщение которое взялось отсюда? $msg = $q->get(0);
Тогда как оно может быть long deliveryTag? (то есть long?)

Что, если брать сообщения через consume?

И еще, у меня была проблема: я вытягивал из очереди 5-10 сообщений, а в очереди сообщений было больше - так вот после consume(5) в очереди сообщений больше не оставалось. Так задумано, что я должен вытянуть единожды сразу все?

Прошу прощения, просто отсутствие подробных доков и обилие примеров с питоновских либ путает меня, и вводит в ступор.

2010/10/21 Alexandre Kalendarev <aka...@gmail.com>

[Alexandre Kalendarev]

21 октября 2010 г. 21:36 пользователь Rauan Maemirov <ra...@maemirov.com> написал:

Сильно ли отличается 0.8 от 0.9.1?
Если использовать сейчас 1.7.2, не возникнет ли никаких проблем при переходе на 2.1?

практически нет,

кажется добавили приаттаченные файлы или что-то типа того

базовые классы не трогали

 

>№ consumerTag (первое сообщение)
Что-то я все никак не врублюсь.
'1' - это сообщение которое взялось отсюда? $msg = $q->get(0);
Тогда как оно может быть long deliveryTag? (то есть long?)

чтоб лучше понять что такое  consumer-tag лучше всего протрейсить сообщения 

http://habrahabr.ru/blogs/personal/74609/

http://habrahabr.ru/blogs/personal/74613/

бо большому счету - это просто номер отправленного сообщения.

 

Что, если брать сообщения через consume?

 

все тоже самое

 

И еще, у меня была проблема: я вытягивал из очереди 5-10 сообщений, а в очереди сообщений было больше - так вот после consume(5) в очереди сообщений больше не оставалось. Так задумано, что я должен вытянуть единожды сразу все?

consume() помечает что все сообщения "выбраны". Чтоб этого не было надо использовать флаг AMQP_NOACK

а вот метод basic.ack как раз по consumer-tag дает знать, какие сообщения выбраны а какие еще нет.

использование consume() предназначено для синхронных систем, и в РНР реализации имеет отрицательные побочные эффеты, ввиде переполнения памяти на больших объемах. По этому этот кусок будет переделываться на подобии http://code.google.com/p/rabbitcpp/ через кэллбэк 

 

 

Прошу прощения, просто отсутствие подробных доков и обилие примеров с питоновских либ путает меня, и вводит в ступор.

питоновская либа одна из самых продвинутых либ, близкой к стандарту AMQP

я старался сделать юзобилити более дружелюбной, без лишних обвязок.

предложения и критика всегда принимаются

Александр 

 

[Rauan Maemirov]

>бо большому счету - это просто номер отправленного сообщения.

Уже понятнее.

Если я запущу десять воркеров, отправлять ack('1') уже не получится ведь? Откуда тогда брать deliveryTag? (в разрезе модуля)

2010/10/22 Alexandre Kalendarev <aka...@gmail.com>

[Alexandre Kalendarev]

21 октября 2010 г. 22:57 пользователь Rauan Maemirov <ra...@maemirov.com> написал:

>бо большому счету - это просто номер отправленного сообщения.

Уже понятнее.

Если я запущу десять воркеров, отправлять ack('1') уже не получится ведь? Откуда тогда брать deliveryTag? (в разрезе модуля)

конечно, deliveryTag индивидуальный для каждого сообщения и берется из $msg = $quene->get();

про consume  не скажу точно, там этот функционал не реализован... 

рекомендую использовать get, он больше подходит для online-задач 

более функционален

трейсить сообщения пробовал, как было написано в статье?

[Rauan Maemirov]

Да, попробовал. Много чего открыл для себя. Но возникла следующая проблема:

1287742141275: conn#11 ch#0 -> {#method<connection.start-ok>(client-properties={},mechanism=PLAIN,response=kiwiGCQQvTmwS2tJ,locale=en_US),null,""}

1287742141276: conn#11 ch#0 <- {#method<connection.start>(version-major=8,version-minor=0,server properties={product=RabbitMQ, information=Licensed under the MPL.  See http://www.rabbitmq.com/, platform=Erlang/OTP, copyright=Copyright (C) 2007-2010 LShift Ltd., Cohesive Financial Technologies LLC., and Rabbit Technologies Ltd., version=1.7.2},mechanisms=PLAIN AMQPLAIN,locales=en_US),null,""}

1287742141316: conn#11 ch#0 <- {#method<connection.tune>(channel-max=0,frame-max=131072,heartbeat=0),null,""}

1287742141357: conn#11 ch#0 -> {#method<connection.tune-ok>(channel-max=0,frame-max=131072,heartbeat=0),null,""}

1287742141357: conn#11 ch#0 -> {#method<connection.open>(virtual-host=/,capabilities=,insist=true),null,""}

1287742141398: conn#11 ch#0 <- {#method<connection.open-ok>(known-hosts=smotrelka:5672),null,""}

1287742141439: conn#11 ch#1 -> {#method<channel.open>(out-of-band=),null,""}

1287742141480: conn#11 ch#1 <- {#method<channel.open-ok>(),null,""}

1287742141523: conn#11 ch#1 -> {#method<basic.publish>(ticket=0,exchange=kiwi.fanout,routing-key=watch.send,mandatory=true,immediate=true),#contentHeader<basic>(content-type=text/plain, content-encoding=null, headers=null, delivery-mode=2, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null),"{"hash":"t4agm7i8h2hm","username":"Quessir","friends":[831,295,75,2735]}"}

1287742141536: conn#11 ch#1 -> {#method<channel.close>(reply-code=200,reply-text=200,class-id=0,method-id=0),null,""}

1287742141565: conn#11 ch#1 <- {#method<basic.return>(reply-code=313,reply-text=not_delivered,exchange=kiwi.fanout,routing-key=watch.send),#contentHeader<basic>(content-type=text/plain, content-encoding=null, headers=null, delivery-mode=2, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null),"{"hash":"t4agm7i8h2hm","username":"Quessir","friends":[831,295,75,2735]}"}

1287742141565: conn#11 ch#1 <- {#method<channel.close-ok>(),null,""}

1287742141609: conn#11 uncaught java.io.EOFException

at java.io.DataInputStream.readUnsignedByte(DataInputStream.java:290)

at com.rabbitmq.client.impl.Frame.readFrom(Frame.java:118)

at com.rabbitmq.tools.Tracer$DirectionHandler.readFrame(Tracer.java:238)

at com.rabbitmq.tools.Tracer$DirectionHandler.doFrame(Tracer.java:272)

at com.rabbitmq.tools.Tracer$DirectionHandler.run(Tracer.java:316)

at java.lang.Thread.run(Thread.java:636)

Причем код выглядит примерно так:

$parms = AMQP_IMMEDIATE | AMQP_MANDATORY;

$properties = array(

'content_type' => 'text/json',

'content_encoding' => 'UTF-8',

'delivery_mode' => 2

);

$this->_exchange = new AMQPExchange($this->getConnection(), $exchangeName);

$this->_exchange->publish($message, $routingKey, $parms, $properties);

$exchangeName - exchange с типом 'fanout'.

Как видите, во-первых, $properties проигнорирован, во-вторых, reply-text=not_delivered

Вопрос: если воркер с очередью, которая привязана к данному exchange, не запущен - сообщение возвращается, как неотправленное?

[Rauan Maemirov]

Корень зла лежал в 

$parms = AMQP_IMMEDIATE | AMQP_MANDATORY;

Объяснение: http://bunnyamqp.wordpress.com/2009/08/21/amqp-basic-publish-immediate-versus-mandatory/

2010/10/22 Rauan Maemirov <ra...@maemirov.com>

[Alexandre Kalendarev]

и еще если exchange имеет тип fanout, то ключ не волспринимается

можно поле оставлять пустым

22 октября 2010 г. 16:48 пользователь Rauan Maemirov <ra...@maemirov.com> написал:

--
Alexandre

[Rauan Maemirov]

Возникла одна проблема:

1). Взяли сообщение из queue. Обработали его.

$carrot = $queue->get();

...

$deliveryTag = $carrot['delivery_tag'];

...

2). Послали другое сообщение в queue2 (fanout exchange)

3). Послали ack с deliveryId от первого действия.

$queue->ack($deliveryTag);

1287841805774: conn#29 ch#1 -> {#method<basic.ack>(delivery-tag=1,multiple=false),null,""}

1287841805814: conn#29 ch#0 <- {#method<connection.close>(reply-code=503,reply-text=COMMAND_INVALID - unknown delivery tag 1,class-id=60,method-id=80),null,""}

Что я пропустил? Если очередь была задекларирована как 

$queue->declare($queueName, AMQP_DURABLE | AMQP_NOACK);

Должен ли я был делать опять же

$exchange->declare($queueName, 'fanout', AMQP_DURABLE | AMQP_NOACK);

Попробывал $queue->get(AMQP_NOACK), но ошибки те же.

Этот трейсер значительно облегчает поиск ошибок.

2010/10/22 Alexandre Kalendarev <aka...@gmail.com>

[Rauan Maemirov]

Проставил AMQP_NOACK везде, где можно - не помолго.

Кстати, в модуле течет память. Попробовать запустить в бесконечный цикл:

Fatal error: Allowed memory size of 16777216 bytes exhausted (tried to allocate 72 bytes) in ...

Call Stack:

    0.0005     644048   1. {main}() ...send.php:0

    2.1739   16719224   2. AMQPQueue->get() ...send.php:33

А может, это из-за ошибки в предыдущем ack.

2010/10/23 Rauan Maemirov <ra...@maemirov.com>

[Alexandre Kalendarev]

да я в курсе

она не течет а накапливается, а реализуется в конце бесконечного цикла (т.е. не реализуется освобождение)

сейчас переделываю эту часть в корне

23 октября 2010 г. 18:13 пользователь Rauan Maemirov <ra...@maemirov.com> написал:

--
Alexandre

[Rauan Maemirov]

Проблема с ack однако осталась.

2010/10/24 Alexandre Kalendarev <aka...@gmail.com>

да я в курсе

она не течет а накапливается, а реализуется в конце бесконечного цикла (т.е. не реализуется освобождение)

сейчас переделываю эту часть в корне

--
Alexandre

[Alexandre Kalendarev]

в чем проблема?

я уже потерял мысль

24 октября 2010 г. 13:20 пользователь Rauan Maemirov <ra...@maemirov.com> написал:

--
Alexandre

[Rauan Maemirov]

Возникла одна проблема:

1). Взяли сообщение из queue. Обработали его.

$carrot = $queue->get();

...

$deliveryTag = $carrot['delivery_tag'];

...

2). Послали другое сообщение в queue2 (fanout exchange)

3). Послали ack с deliveryId от первого действия.

$queue->ack($deliveryTag);

1287841805774: conn#29 ch#1 -> {#method<basic.ack>(delivery-tag=1,multiple=false),null,""}

1287841805814: conn#29 ch#0 <- {#method<connection.close>(reply-code=503,reply-text=COMMAND_INVALID - unknown delivery tag 1,class-id=60,method-id=80),null,""}

Что я пропустил? Если очередь была задекларирована как 

$queue->declare($queueName, AMQP_DURABLE | AMQP_NOACK);

Должен ли я был делать опять же

$exchange->declare($queueName, 'fanout', AMQP_DURABLE | AMQP_NOACK);

Попробывал $queue->get(AMQP_NOACK), но ошибки те же.

2010/10/25 Alexandre Kalendarev <aka...@gmail.com>

[Alexandre Kalendarev]

ОК

я эту ситуацию проверю.

Одна из возможных причин - это очередь может быть с флагом AUTO_DELETE (default)

по этому когда из нее выбирают последнее сообщение - она удаляется

хотя судя по ошибке eply-code=503,reply-text=COMMAND_INVALID - unknown delivery tag 1,class-id=60,method-id=80

ошибка возникает в методе ACK

вот что пишет Протокол

This method acknowledges one or more messages delivered via the Deliver or GetOk

methods. The client can ask to confirm a single message or a set of messages up to and including a specific message

Ок

я проверю чуть позже

Александр

24 октября 2010 г. 22:27 пользователь Rauan Maemirov <ra...@maemirov.com> написал:

--
Alexandre

[Alexandre Kalendarev]

23 октября 2010 г. 18:13 пользователь Rauan Maemirov <ra...@maemirov.com> написал:

Проставил AMQP_NOACK везде, где можно - не помолго.

Кстати, в модуле течет память. Попробовать запустить в бесконечный цикл:

Fatal error: Allowed memory size of 16777216 bytes exhausted (tried to allocate 72 bytes) in ...

Call Stack:

    0.0005     644048   1. {main}() ...send.php:0

    2.1739   16719224   2. AMQPQueue->get() ...send.php:33

эти 72 байта "текут" из rabbitmq-c насколько я припоминаю, когда я чесал валгриндом,

почему он их ловит - я не знаю, наверно это связано с системными вызовами.

теперь я разрабатываю под маком - там нет такой классной утилиты, приходится использовать leaks

а она хуже, хотя тоже что-то ловит.

сейчас я сделал новый механизм распределения памяти,

нужно протестить на предмет утечек длинных сообщений (послать длинное сообщение -  более 131К потом короткое потом еще длиннее  2 х 131К)

да и вообще сделал другой механизм - через кэллбэки

если есть желание помочь в тестировании чуть позже выложу рабочую версию

очевидно отдельным тредом

Alexandre

[Rauan Maemirov]

Само собой, выкладывайте. Проблему с ack воспроизвести не получилось?

2010/10/26 Alexandre Kalendarev <aka...@gmail.com>

[Alexandre Kalendarev]

26 октября 2010 г. 11:16 пользователь Rauan Maemirov <ra...@maemirov.com> написал:

Само собой, выкладывайте. Проблему с ack воспроизвести не получилось?

не воспроизводил еще 

пока хочу довести до ума первое дело

Alexandre