[Feature] Bulk Consumer #413

devrck · 2016-12-07T17:54:30Z

#403 implementation

This new type of consumer gets x messages from the queue in unacknowledged state
Executes for each of them the @execute() method
When the batch is complete or it's timeout wait is fulfilled then it will fire the method @batchExecute()
If the @batchExecute() is without any exceptions it will acknowledged the messages if not it will re-queue them to be handled by another worker

[ Obs ]: Message can get stuck on the queue forever because of the behavior of the @batchExecute() when it has exceptions. I am open for suggestions on how to fix this.

Configuration used in app/config/config.yml

# RabbitMQ Configuration
old_sound_rabbit_mq:
    connections:
        default:
            host:     '%rabbit_mq_conn_default_host%'
            port:     '%rabbit_mq_conn_default_port%'
            user:     '%rabbit_mq_conn_default_user%'
            password: '%rabbit_mq_conn_default_password%'
            vhost:    '%rabbit_mq_conn_default_vhost%'
            lazy:     '%rabbit_mq_conn_default_lazy%'
    batch_consumers:
        devck_basic:
            connection:       default
            exchange_options: {name: '%rabbit_mq_exchange_devck%', type: fanout}
            queue_options:    {name: '%rabbit_mq_queue_devck%'}
            callback:         devck.basic
            qos_options:      {prefetch_size: 0, prefetch_count: 2, global: false}
            timeout_wait:     5
            auto_setup_fabric: false
            idle_timeout_exit_code: -2

Callback class

<?php

namespace AppBundle\Service;

use OldSound\RabbitMqBundle\RabbitMq\BatchConsumerInterface;
use PhpAmqpLib\Message\AMQPMessage;

class DevckBasicConsumer implements BatchConsumerInterface
{
    public function batchExecute()
    {
        echo 'Doing batch execution', PHP_EOL;
    }

    /**
     * @inheritDoc
     */
    public function execute(AMQPMessage $msg)
    {
        echo $msg->body, PHP_EOL;

        return true;
    }
}

…ssing

skafandri · 2016-12-07T18:29:33Z

How I access the messages in batchExecute() ?

skafandri · 2016-12-07T18:31:25Z

RabbitMq/BatchConsumerInterface.php

+interface BatchConsumerInterface extends ConsumerInterface
+{
+    public function batchExecute();
+}


new line :)

skafandri · 2016-12-07T18:31:32Z

RabbitMq/BatchConsumer.php

+    {
+        $this->getChannel()->wait(null, false, $timeout);
+    }
+}


new line :)

skafandri · 2016-12-07T18:31:43Z

Command/BatchConsumerCommand.php

+    {
+        return 'old_sound_rabbit_mq.%s_batch';
+    }
+}


new line :)

devrck · 2016-12-08T15:12:51Z

Well my premise is that you do work in the execute() part and then in the batchExecute() you will do a action that is very costly.

Take by example a batch process for inserting in a database using Doctrine. You do the work in execute(), persisting objects, modifying objects and then in the batchExecute() you will do a flush.

After you've asked how do you access the messages I thought you should have access and I can give access to the messages in the batchExecute() modifying its signature to:

/**
 * @param   AMQPMessage[]  $messages
 *
 * @return   void
 */
public function batchExecute(array $messages);

Q: @skafandri what do you think?

PS: I will fix the CR 😄 as soon as possible.

skafandri · 2016-12-08T15:37:44Z

I ma not sure if I understand the bulk idea in this context well. What I understood is:

Get a batch of messages
Do some work
Return true to ack them all or false to nack them all
It should be impossible (as good as we can do) to ack/nack only some messages

devrck · 2016-12-09T11:15:37Z

Yes you've got it.

The problem with this is that if your message generates an exception it will be stuck on your queue forever if you don't have logic to get it off. I can implement a mechanism to know if your messages has been passed along for some time but I don't know if I should do this or the developer.

Whats your opinion?

skafandri · 2016-12-09T11:46:47Z

I think about it to be similar to an SQL transaction

start transaction;
insert into order_line(`id`, `product_id`) values (1, 10);
insert into order_line(`id`, `product_id`) values (1, 11);
insert into order_line(`id`, `product_id`) values (1, NULL);
commit;

As long as product_id is not nullable, all the inserts will fail.
If the bulk operation is not atomic then I wouldn't really need this feature. All I would do is:

public function execute(AMQPMessage $msg)
{
           $result = $this->processMessage($msg);
          
           if ($this->reachedBulkCondition()) {
                       $this->bulkAction();
                       $this->resetBulkCondition();
           }
           return $result;
}

Where reachedBulkCondition can check for the number of processed messages or an elapsed time or both.

Now back to your implementation, if batchExecute throws an exception, I think ALL the messages should be nacked. If one of those messages is causing the exception it will effectively get stuck in the queue. But this is the developer's job to handle this situation. The same behavior is present with the single consumer as well.

In other words, I expect batchExecute to behave exactly like execute. The only difference being the unit of work, AMQPMessage for execute and AMQPMessage[] for batchExecute.

I wouldn't mind creating an AMQPBulkMessage which will contain a collection of AMQPMessages to make it clear to the user that he is dealing (nacking/acking) with a collection of messages.

devrck · 2016-12-09T14:04:32Z

Hmm now that I think about my solution is a work around to this batchExecute(array $messages) and the existence of execute(AMQPMessage $msg) doesn't seem of any use. I will give another shoot at this and get all the messages in the batchExecute(array $messages) part.

…ause it does the work in batches, could be added later on maybe, but with other meaning (how many batches not how many messages)

… messages and it can return for each message the response ot ack/nack/requeue/reject or 1 single response applied for all.

…consumer just has an array of messages instead of a single one

…nd then pass it to the callback

devrck · 2016-12-10T20:08:30Z

Fixed CR & implemented another approach to the bulk consumer.

devrck · 2016-12-16T05:35:26Z

Hi guys did anybody got a chance to look over this?

devrck · 2016-12-16T05:47:44Z

A consumer can look like this.

It basically has the same structure as before but now i pass an array of messages to batchExecute (as @skafandri 👍 pointed out that it should work like execute(AMQPMessage $msg) and be atomic).

In the first example you can just return one bool value that is the same as saying the same value for each message:

namespace AppBundle\Service;

use OldSound\RabbitMqBundle\RabbitMq\BatchConsumerInterface;
use PhpAmqpLib\Message\AMQPMessage;

class DevckBasicConsumer implements BatchConsumerInterface
{
    /**
     * @inheritDoc
     */
    public function batchExecute(array $messages)
    {
        echo sprintf('Doing batch execution%s', PHP_EOL);
        foreach ($messages as $message) {
            $this->executeSomeLogicPerMessage($message);
        }

        return true;
    }
}

The second example offers you more control on what to do with each individual message but you have to keep track of the messages and assign for each delivery_tag a value and then return that result back to the batch consumer:

namespace AppBundle\Service;

use OldSound\RabbitMqBundle\RabbitMq\BatchConsumerInterface;
use PhpAmqpLib\Message\AMQPMessage;

class DevckBasicConsumer implements BatchConsumerInterface
{
    /**
     * @inheritDoc
     */
    public function batchExecute(array $messages)
    {
        echo sprintf('Doing batch execution%s', PHP_EOL);
        $result = [];
        /** @var AMQPMessage $message */
        foreach ($messages as $message) {
            $result[(int)$message->delivery_info['delivery_tag']] = $this->executeSomeLogicPerMessage($message);
        }

        return $result;
    }
}

@skafandri doing another class AMQPBulkMessages is the best choice IMHO for this batch consumer but from what I've investigated (maybe I need more time) you still get one message and then the next.

skafandri · 2016-12-16T06:12:43Z

Looks good for me. I think we should document some edge cases so users won't get confused.

What happens of the batch size is smaller than the prefetch count?
What happens if there are less messages available in the queue than the batch size.
We should also mention that a batch is not fixed. With a batch of 3, if I nack [m1, m2, m3] there is no guarantee (most likely not) that another consumer will receive the same batch, it may receive [m1, m2, m5]

devrck · 2016-12-22T06:57:18Z

First of all this is the configuration for a batch consumer (It the same as above):

  # RabbitMQ Configuration
old_sound_rabbit_mq:
    connections:
        default:
            host:     '%rabbit_mq_conn_default_host%'
            port:     '%rabbit_mq_conn_default_port%'
            user:     '%rabbit_mq_conn_default_user%'
            password: '%rabbit_mq_conn_default_password%'
            vhost:    '%rabbit_mq_conn_default_vhost%'
            lazy:     '%rabbit_mq_conn_default_lazy%'
    batch_consumers:
        devck_basic:
            connection:       default
            exchange_options: {name: '%rabbit_mq_exchange_devck%', type: fanout}
            queue_options:    {name: '%rabbit_mq_queue_devck%'}
            callback:         devck.basic
            qos_options:      {prefetch_size: 0, prefetch_count: 2, global: false}
            timeout_wait:     5
            auto_setup_fabric: false
            idle_timeout_exit_code: -2

Getting back to your questions:

Q: What happens of the batch size is smaller than the prefetch count?
A: The prefetch_count is the batch_size by my implementation, and I think I have to restrict user to input a value greater or equal to 2 ( I will check and push changes if I haven't yet).

Q: What happens if there are less messages available in the queue than the batch size?
A: It will get all the messages it has and after timeout_wait of no more messages received on the line you will receive the messages in the batchExecute() method. timeout_wait is different from idle_timeout. timeout_wait acts like a buffer in case something comes along on the line.

Q: We should also mention that a batch is not fixed. With a batch of 3, if I nack [m1, m2, m3] there is no guarantee (most likely not) that another consumer will receive the same batch, it may receive [m1, m2, m5]
A: I agree there is no guarantee that you will receive the same batch because the dispatching is handled by the RabbitMq server.

Additional situations:

Prefetch with large messages:

- Queue: 1 messages of 3kb, N messages of 1kb
- Qos: prefetch_size: 3kb , prefetch_count: 3
- In this scenario how does the system behave? (It should receive the first messages of 3kb and send it to the batchExecute() because it thats the maximum and it should not stick around timeout_wait time for other messages).
- Queue: 1 messages of 3kb, N messages of 1kb
- Qos: prefetch_size: 2kb, prefetch_count: 3
- In this scenario I don't see a solution, that 1st message will get stuck on the queue forever correct?

skafandri · 2016-12-22T07:21:42Z

good job, let's add this to README.md so we can review it better to make sure is accurate, helpful, and fits well with the current documentation.

Nightbr · 2017-02-21T08:42:16Z

Any news? It is a great feature !! Awesome works! We really need it!

devrck · 2017-02-21T09:28:35Z

@Nightbr will get back to it to complete the documentation and add revise the code.

christianmeller · 2017-03-20T08:40:30Z

Thank you for this awesome work. This is exactly what we was looking for.
Could you estimate how long it should take to finish this feature? We really need it as well ;-)

devrck · 2017-06-06T16:07:08Z

@skafandri after a long time I've added the README part.

stloyd · 2017-06-07T09:50:32Z

README.md

+    }
+}
+
+```


Should be: "```php"

stloyd · 2017-06-07T09:52:03Z

README.md

+        // you ack all messages got in batch
+        return true; 
+    }
+}


Missing close tag: "```"

stloyd · 2017-06-07T10:15:08Z

RabbitMq/BatchConsumer.php

+
+        $this->maybeStopConsumer();
+
+        if (!is_null($this->getMemoryLimit()) && $this->isRamAlmostOverloaded()) {


stloyd · 2017-06-07T10:15:43Z

RabbitMq/BatchConsumer.php

+        $this->consumed++;
+        $this->maybeStopConsumer();
+
+        if (!is_null($this->getMemoryLimit()) && $this->isRamAlmostOverloaded()) {


stloyd · 2017-06-07T10:17:38Z

RabbitMq/BatchConsumer.php

+    /**
+     * @var int
+     */
+    protected $consumed = 0;


If this is not expected to be extended please mark all variables as private.

stloyd · 2017-06-07T10:18:40Z

DependencyInjection/Configuration.php

    public function getConfigTreeBuilder()
    {
        $tree = new TreeBuilder();

-        $rootNode = $tree->root('old_sound_rabbit_mq');
+        $rootNode = $tree->root($this->name);


Why? We don't plan to change the configuration name.

Make use of already defined alias.

stloyd · 2017-06-07T10:18:53Z

Command/BatchConsumerCommand.php

+        $this->consumer = $this->getContainer()
+            ->get(sprintf($this->getConsumerService(), $input->getArgument('name')));
+
+        if (!is_null($input->getOption('memory-limit')) && ctype_digit((string) $input->getOption('memory-limit')) && $input->getOption('memory-limit') > 0) {


stloyd · 2017-06-07T10:19:42Z

Command/BatchConsumerCommand.php

+
+    public function restartConsumer()
+    {
+        // TODO: Implement restarting of consumer


This should be removed or added before merging.

stloyd · 2017-06-07T10:20:09Z

Command/BatchConsumerCommand.php

+    protected $consumer;
+
+    /**
+     * @return  void


Such phpdoc is useless IMO.

devrck · 2017-06-07T14:39:24Z

Waiting review 😄

christianmeller · 2017-06-07T15:34:43Z

DependencyInjection/OldSoundRabbitMqExtension.php

+                    $consumer['qos_options']['global']
+                ))
+            ;
+


method call for setIdleTimeoutExitCode (key: idle_timeout_exit_code) is missing

stloyd · 2017-06-08T06:40:04Z

RabbitMq/BatchConsumer.php

+                $this->batchConsume();
+            }
+
+            $timeoutWanted = ($isConsuming) ? $this->getTimeoutWait() : $this->getIdleTimeout();


( & ) are useless here.

stloyd · 2017-06-08T06:40:16Z

RabbitMq/BatchConsumer.php

+
+    public function batchConsume()
+    {
+        if ($this->batchCounter == 0) {


0 === $this->batchCounter

Implemented.

stloyd · 2017-06-08T06:41:37Z

RabbitMq/BatchConsumer.php

+
+        if ($this->forceStop) {
+            $this->stopConsuming();
+        } else {


This whole } else { & return; is useless.

stloyd · 2017-06-08T07:23:49Z

Command/BatchConsumerCommand.php

+
+            pcntl_signal(SIGTERM, array(&$this, 'stopConsumer'));
+            pcntl_signal(SIGINT, array(&$this, 'stopConsumer'));
+//            pcntl_signal(SIGHUP, array(&$this, 'restartConsumer'));


This looks like unused.

stloyd · 2017-06-08T07:30:07Z

RabbitMq/BatchConsumer.php

+                )
+            ));
+            throw $e;
+        } finally {


We cannot use finally as we support PHP 5.3+

Sure, fixed.

…ivate

devrck · 2017-06-12T06:57:51Z

@stloyd what about this PR? What changes do I have to make?

stloyd · 2017-06-12T07:05:15Z

Thank you @devrck !

Bogdan Rancichi added 12 commits December 7, 2016 15:41

updated doc comments for method

04404f9

added schema onfiguration for batch_consumer

27b01a4

send the configuration name from the extension

39951b1

set the configuration name & load batch consumers build

c5b332e

added batch_consumer class definition

2c173a7

initial batch consumer class implementation

1d59ce9

complete the build of batch consumers from extensions

80392e1

added queue configuration key for configuration

e256558

keep track of messages and fix implementation for isCompleteBatch

16d2ec8

created separate command for batch consumers

1cf94dc

created new interface to signal that a callback will have batch proce…

1381d17

…ssing

keep only channel & delivery tag on message

d00bd48

skafandri reviewed Dec 7, 2016

View reviewed changes

RabbitMq/BatchConsumer.php Outdated

{

$this->getChannel()->wait(null, false, $timeout);

}

}

Copy link

Collaborator

skafandri Dec 7, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

new line :)

skafandri reviewed Dec 7, 2016

View reviewed changes

Command/BatchConsumerCommand.php

{

return 'old_sound_rabbit_mq.%s_batch';

}

}

Copy link

Collaborator

skafandri Dec 7, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

new line :)

Bogdan Rancichi added 5 commits December 10, 2016 15:36

refactored BatchConsumerCommand it can not accept --messages flag bec…

8ee3028

…ause it does the work in batches, could be added later on maybe, but with other meaning (how many batches not how many messages)

refactored the meaning of a batch consumer it now accepts an array of…

5521707

… messages and it can return for each message the response ot ack/nack/requeue/reject or 1 single response applied for all.

prevent CR comment about new line.

bbbac11

batch consumer will have just 1 callback that behaves like a regular …

c103721

…consumer just has an array of messages instead of a single one

refactored BatchConsumer implementation to collect all the messages a…

db49a24

…nd then pass it to the callback

devrck and others added 2 commits May 24, 2017 17:39

Merge branch 'master' into batch-consumer

58617ac

[README] updated readme for batch consumer

599b675

stloyd requested changes Jun 7, 2017

View reviewed changes

[change-req] implemented change requests

938d944

christianmeller reviewed Jun 7, 2017

View reviewed changes

[fix] added method call to idletimeout_exit_code

3e0faa3

stloyd reviewed Jun 8, 2017

View reviewed changes

[fixed] implemented suggested code review

6046a41

stloyd requested changes Jun 8, 2017

View reviewed changes

[fix] remove unused call; remove finally clause; made some methods pr…

a39a2a0

…ivate

stloyd approved these changes Jun 12, 2017

View reviewed changes

stloyd merged commit 564e87c into php-amqplib:master Jun 12, 2017

+                  }
+              }
+              ```


		$this->maybeStopConsumer();

		if (!is_null($this->getMemoryLimit()) && $this->isRamAlmostOverloaded()) {

[Feature] Bulk Consumer #413

[Feature] Bulk Consumer #413

Uh oh!

Conversation

devrck commented Dec 7, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

skafandri commented Dec 7, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

devrck commented Dec 8, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

skafandri commented Dec 8, 2016

Uh oh!

devrck commented Dec 9, 2016

Uh oh!

skafandri commented Dec 9, 2016

Uh oh!

devrck commented Dec 9, 2016

Uh oh!

devrck commented Dec 10, 2016

Uh oh!

devrck commented Dec 16, 2016

Uh oh!

devrck commented Dec 16, 2016

Uh oh!

skafandri commented Dec 16, 2016

Uh oh!

devrck commented Dec 22, 2016

Uh oh!

skafandri commented Dec 22, 2016

Uh oh!

Nightbr commented Feb 21, 2017

Uh oh!

devrck commented Feb 21, 2017

Uh oh!

christianmeller commented Mar 20, 2017

Uh oh!

devrck commented Jun 6, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

devrck commented Dec 7, 2016 •

edited

Loading

devrck commented Dec 8, 2016 •

edited

Loading