Using redis as a queue

[NotionCommotion]

I wish to use redis as a queue.  The way I envision it working is an ajax call to a PHP script will push something on the queue, and then another PHP script will have a blocking command (i.e. http://redis.io/commands/brpoplpush) to receive the something, and deal with it.

 

I've installed it as shown below.  Note that I have not edited /etc/redis.conf or /etc/php.ini, nor installed any additional sorts of redis clients.

yum remove redis.x86_64
yum install redis32u.x86_64
cat /etc/redis.conf
yum install php56u-pecl-redis.x86_64 # Extension for communicating with the Redis key-value store
yum install php56u-pecl-redis-debuginfo.x86_64 # Debug information for package php56u-pecl-redis
yum install redis32u-debuginfo.x86_64 # Debug information for package redis32u
#Requires tcl (already installed)
chkconfig --add redis
chkconfig --level 345 redis on

phpinfo() shows redis 2.2.8as being installed.  Why not redis3.2?

 

Do I need a redis client?  For instance, one of http://redis.io/clients#php?  If so, which one?

 

I've tried $redis = new Redis();, however, the class doesn't exist.

 

What next?

 

Thanks

[kicken]

phpinfo() shows redis 2.2.8as being installed.  Why not redis3.2?

It's showing 2.2.8 because that is the version of the redis client you installed.

 

Do I need a redis client?  For instance, one of http://redis.io/clients#php?  If so, which one?

You did that when you installed the php56u-pecl-redis.x86_64 package. Documentation is on Github for how to use it.

 

I've tried $redis = new Redis();, however, the class doesn't exist.

If it installed correctly then according to the documentation that should work. You may need to restart your server and/or PHP for the change to take effect.

[NotionCommotion]

Thanks Kicken!

 

and not 3.2 based on doing: yum install redis32u.x86_64?

 

Also, yes, I see how I did in fact include a client.

 

Turns out I needed to only add the following to php.ini.  Probably questions later, but am now talking!

extension = redis.so

[NotionCommotion]

Okay, I add something to a queue

<?php
    // Ajax file
   //Connecting to Redis server on localhost
   $redis = new Redis();
   $redis->connect('127.0.0.1', 6379);
   $rs=$redis->rPush('stack', 'Hello!');
   var_dump($rs);

Now need to get it back.  Reading http://redis.io/commands/rpoplpush, I don't wish to just pop it off as it is not reliable.  Instead, I should use rpoplpush (or brpoplpush if I want it to be a blocking function which I do).  So, I did the following and it appears to work.

<?php
   //Consumer
   //Connecting to Redis server on localhost
   $redis = new Redis();
   $redis->connect('127.0.0.1', 6379);
   $rs=$redis->bRPopLPush('stack','stack2',0);
   var_dump($rs);

It is my understanding that the first argument to bRPopLPush is the variable name, and the last is the timeout time and I am thinking that 0 is indefinite.  Correct?

 

What is the second argument all about?

[NotionCommotion]

Any recommendations on the proper way to implement a queue?  Thanks

[requinix]

As the docs say, brpoplpush - which you can read as a "b"locking "r"ight "pop" and "l"eft "push" - deals with two lists, and moves one element from one to the other. If all you need is a simple queue with blocking then do lpush+brpop or rpush+blpop (whichever "direction" makes most sense to you).

[NotionCommotion]

As the docs say, brpoplpush - which you can read as a "b"locking "r"ight "pop" and "l"eft "push" - deals with two lists, and moves one element from one to the other. If all you need is a simple queue with blocking then do lpush+brpop or rpush+blpop (whichever "direction" makes most sense to you).

 

Okay, that is what I thought.  Just move an element from one list to another.  But why?  Why not just pop it off a list and deal with it?  Because it gets popped off, something goes wrong, and it is lost?  That's not good, but what does moving it to another list help?

 

Stop.  I better read your response again.  So, don't bother with brpoplpush, and just push and pop.  Okay, I am good with that, but why does the documentation say it is not reliable to do so?  Or am I misreading it?

[kicken]

But why?  Why not just pop it off a list and deal with it?  Because it gets popped off, something goes wrong, and it is lost?  That's not good, but what does moving it to another list help?

The documentation for rpoplpush explains some.

However in this context the obtained queue is not reliable as messages can be lost, for example in the case there is a network problem or if the consumer crashes just after the message is received but it is still to process.

RPOPLPUSH (or BRPOPLPUSH for the blocking variant) offers a way to avoid this problem: the consumer fetches the message and at the same time pushes it into a processing list.

If you just did the basic approach of push/pop from a single stack then the item could be lost without being processed if something were to happen during processing.

while (null !== $rs=$redis->lpop('queue')){
   DoStuff($rs);
}

Say DoStuff throws exception or the power fails in the middle of it. That item will have been removed from the queue but not successfully processed.

 

 

By pushing the item into another list you can remove it from the queue and process it without risking loosing it. For example:

while (null !== $rs=$redis->bRPopLPush('queue','processing',0)){
   DoStuff($rs);
   $redis->lrem('processing', 1,$rs);
}

Here the item is removed from the queue then added to a processing list. The code then does what it needs to do and finally removes it from the processing list.

 

If there were a problem or power failure the item is preserved in the processing list. You'd then have another process that periodically checks the processing list for items past a certain age and restore them to the queue so they can be re-processed.

[NotionCommotion]

Thanks kicken,  Very nice explanation.

 

What if I added to the queue "pay kicken $1,000".  I move it from one queue to another, process it, and then BOOM, I lose power.  How would I prevent sending you an additional $1,000?

[NotionCommotion]

The while statement didn't seem to work, and the argument order in lrem needed to be changed.

 

I am still getting quirky results.   Does anything look wrong below?

            $redis = new \Redis();
            $redis->connect('127.0.0.1', 6379);
            return $redis->lPush('queue_'.$guid, json_encode($data));

    $id=$_GET['guid'];
    //Connecting to Redis server on localhost
    $redis = new Redis();
    $redis->connect('127.0.0.1', 6379);
    if($rs=$redis->rpop("processing_$id")) {
        //Something was in the queue, but didn't get processed
        $rsp=[$rs,200];
    }
    else {
        $rsp=[null,204];  //Maybe something else?
        //while (null !== $rs=$redis->bRPopLPush("queue_$id","processing_$id",300)){    //Causes error?
        if($rs=$redis->bRPopLPush("queue_$id","processing_$id",300)) {
            $rsp=[$rs,200];
            //What if something goes wrong before removed from list?
            $redis->lrem("processing_$id", $rs, 1);  //Note that order of arguments is different with redis and phpredis!
        }
    }

[NotionCommotion]

Trying to make things simpler until I get it working right, I've got rid of bRPopLPush, and am simply using $rs=$redis->brpop("queue_$id",300);.

 

Every minute, I get the following error which occurs on that line.

 

Uncaught exception 'RedisException' with message 'read error on connection'

 

One solution I read was to do echo 1 > /proc/sys/vm/overcommit_memory, however, it doesn't work for me.

 

I've read elsewhere to disable persistent connections for phpredis, but I don't know how to do so.

 

Any recommendations?  Thanks

Edited October 30, 2016 by NotionCommotion

[kicken]

What if I added to the queue "pay kicken $1,000".  I move it from one queue to another, process it, and then BOOM, I lose power.  How would I prevent sending you an additional $1,000?

You'd have to see what tools you have available from your payment system. For example one could do a request to begin the transaction and get a payment transaction ID then update your database with that ID. After that commit the payment transaction and update your db. Finally remove the item from the processing queue.

 

If there is a failure and the item is re-added to the queue then when it's processed again you can first check the database for the transaction ID. If one exists, query the payment provider to find the state of that transaction. If it is still pending then try and commit it again. If it's complete update your DB. If it's not found either try again or send the item out to a person for further investigation and manual reconciliation.

 

Every minute, I get the following error which occurs on that line.

I'd check PHP's default_socket_timeout setting and try increasing it. It defaults to 60 seconds so that seems to line up with your observation.

[NotionCommotion]

I'd check PHP's default_socket_timeout setting and try increasing it. It defaults to 60 seconds so that seems to line up with your observation.

 

I didn't expect this to help as I thought php56u-pecl-redis.x86_64 might be independent of the php.ini settings, but it worked!

[requinix]

I thought php56u-pecl-redis.x86_64 might be independent of the php.ini settings

:/ It's running from within PHP so why would it be not be affected by PHP settings?