redis驱动支持断线重连

Author: yunwuxin

src/queue/connector/Database.php

@@ -65,7 +65,8 @@ public static function __make(Db $db, $config)
 
     public function size($queue = null)
     {
-        return $this->db->name($this->table)
+        return $this->db
+            ->name($this->table)
             ->where('queue', $this->getQueue($queue))
             ->count();
     }
@@ -170,7 +171,7 @@ protected function getNextAvailableJob($queue)
             ->where(function (Query $query) {
                 $query->where(function (Query $query) {
                     $query->whereNull('reserve_time')
-                        ->where('available_time', '<=', $this->currentTime());
+                          ->where('available_time', '<=', $this->currentTime());
                 });
 
                 //超时任务重试

src/queue/connector/Redis.php

@@ -13,6 +13,7 @@
 
 use Closure;
 use Exception;
+use RedisException;
 use think\helper\Str;
 use think\queue\Connector;
 use think\queue\InteractsWithTime;
@@ -46,7 +47,7 @@ class Redis extends Connector
      */
     protected $blockFor = null;
 
-    public function __construct(\Redis $redis, $default = 'default', $retryAfter = 60, $blockFor = null)
+    public function __construct($redis, $default = 'default', $retryAfter = 60, $blockFor = null)
     {
         $this->redis      = $redis;
         $this->default    = $default;
@@ -60,18 +61,47 @@ public static function __make($config)
             throw new Exception('redis扩展未安装');
         }
 
-        $func = $config['persistent'] ? 'pconnect' : 'connect';
+        $redis = new class($config) {
+            protected $config;
+            protected $client;
 
-        $redis = new \Redis;
-        $redis->$func($config['host'], $config['port'], $config['timeout']);
+            public function __construct($config)
+            {
+                $this->config = $config;
+                $this->client = $this->createClient();
+            }
 
-        if ('' != $config['password']) {
-            $redis->auth($config['password']);
-        }
+            protected function createClient()
+            {
+                $config = $this->config;
+                $func   = $config['persistent'] ? 'pconnect' : 'connect';
 
-        if (0 != $config['select']) {
-            $redis->select($config['select']);
-        }
+                $client = new \Redis;
+                $client->$func($config['host'], $config['port'], $config['timeout']);
+
+                if ('' != $config['password']) {
+                    $client->auth($config['password']);
+                }
+
+                if (0 != $config['select']) {
+                    $client->select($config['select']);
+                }
+                return $client;
+            }
+
+            public function __call($name, $arguments)
+            {
+                try {
+                    return call_user_func_array([$this->client, $name], $arguments);
+                } catch (RedisException $e) {
+                    if (Str::contains($e->getMessage(), 'went away')) {
+                        $this->client = $this->createClient();
+                    }
+
+                    throw $e;
+                }
+            }
+        };
 
         return new self($redis, $config['queue'], $config['retry_after'] ?? 60, $config['block_for'] ?? null);
     }