fix #173

src/drivers/redis/Queue.php

@@ -12,6 +12,7 @@
 use yii\di\Instance;
 use yii\queue\cli\Queue as CliQueue;
 use yii\redis\Connection;
+use yii\redis\Mutex;
 
 /**
  * Redis Queue.
@@ -24,10 +25,25 @@ class Queue extends CliQueue
      * @var Connection|array|string
      */
     public $redis = 'redis';
+
+    /**
+     * @var Mutex|array|string
+     */
+    public $mutex = [
+        'class' => Mutex::class,
+        'redis' => 'redis',
+    ];
+
+    /**
+     * @var integer
+     */
+    public $mutexTimeout = 3;
+
     /**
      * @var string
      */
     public $channel = 'queue';
+
     /**
      * @var string command class name
      */
@@ -41,6 +57,7 @@ public function init()
     {
         parent::init();
         $this->redis = Instance::ensure($this->redis, Connection::class);
+        $this->mutex = Instance::ensure($this->mutex, Mutex::class);
     }
 
     /**
@@ -56,15 +73,22 @@ public function run($repeat, $timeout = 0)
     {
         return $this->runWorker(function (callable $canContinue) use ($repeat, $timeout) {
             while ($canContinue()) {
-                if (($payload = $this->reserve($timeout)) !== null) {
-                    list($id, $message, $ttr, $attempt) = $payload;
-                    if ($this->handleMessage($id, $message, $ttr, $attempt)) {
-                        $this->delete($id);
+                if ($this->acquire()) {
+                    if (($payload = $this->reserve($timeout)) !== null) {
+                        list($id, $message, $ttr, $attempt) = $payload;
+                        if ($this->handleMessage($id, $message, $ttr, $attempt)) {
+                            $this->delete($id);
+                        }
+
+                    } elseif (!$repeat) {
+                        break;
                     }
-                } elseif (!$repeat) {
-                    break;
+
+                    $this->release();
                 }
             }
+
+            $this->release();
         });
     }
 
@@ -95,10 +119,15 @@ public function status($id)
      */
     public function clear()
     {
-        while (!$this->redis->set("$this->channel.moving_lock", true, 'NX')) {
+        while (!$this->acquire(0)) {
             usleep(10000);
         }
-        $this->redis->executeCommand('DEL', $this->redis->keys("$this->channel.*"));
+
+        try {
+            $this->redis->executeCommand('DEL', $this->redis->keys("$this->channel.*"));
+        } finally {
+            $this->release();
+        }
     }
 
     /**
@@ -110,19 +139,25 @@ public function clear()
      */
     public function remove($id)
     {
-        while (!$this->redis->set("$this->channel.moving_lock", true, 'NX', 'EX', 1)) {
+        while (!$this->acquire(0)) {
             usleep(10000);
         }
-        if ($this->redis->hdel("$this->channel.messages", $id)) {
-            $this->redis->zrem("$this->channel.delayed", $id);
-            $this->redis->zrem("$this->channel.reserved", $id);
-            $this->redis->lrem("$this->channel.waiting", 0, $id);
-            $this->redis->hdel("$this->channel.attempts", $id);
 
-            return true;
-        }
+        try {
+            if ($this->redis->hdel("$this->channel.messages", $id)) {
+                $this->redis->zrem("$this->channel.delayed", $id);
+                $this->redis->zrem("$this->channel.reserved", $id);
+                $this->redis->lrem("$this->channel.waiting", 0, $id);
+                $this->redis->hdel("$this->channel.attempts", $id);
+
+                return true;
+            }
+
+            return false;
 
-        return false;
+        } finally {
+            $this->release();
+        }
     }
 
     /**
@@ -131,11 +166,9 @@ public function remove($id)
      */
     protected function reserve($timeout)
     {
-        // Moves delayed and reserved jobs into waiting list with lock for one second
-        if ($this->redis->set("$this->channel.moving_lock", true, 'NX', 'EX', 1)) {
-            $this->moveExpired("$this->channel.delayed");
-            $this->moveExpired("$this->channel.reserved");
-        }
+        // Moves delayed and reserved jobs into waiting list
+        $this->moveExpired("$this->channel.delayed");
+        $this->moveExpired("$this->channel.reserved");
 
         // Find a new waiting message
         $id = null;
@@ -201,4 +234,27 @@ protected function pushMessage($message, $ttr, $delay, $priority)
 
         return $id;
     }
+
+    /**
+     * Acquire the lock.
+     *
+     * @return boolean
+     */
+    protected function acquire($timeout = null)
+    {
+        $timeout = $timeout !== null ? $timeout : $this->mutexTimeout;
+
+        return $this->mutex->acquire(__CLASS__ . $this->channel, $timeout);
+    }
+
+    /**
+     * Release the lock.
+     *
+     * @return boolean
+     */
+    protected function release()
+    {
+        return $this->mutex->release(__CLASS__ . $this->channel);
+    }
+
 }