Merge branch '3.x' of https://github.com/easy-swoole/queue into 3.x

Author: kiss291323003

README.md

@@ -48,4 +48,22 @@ $job->setWaitConfirmTime(5);//如果5秒内没确认任务，会重新回到队
 $queue->producer()->push($job);//投递任务
 //确认一个任务
 $queue->consumer()->confirm($job);
-```
+```
+
+## 消费者控制
+
+当队列未`pop`到`job`时：
+
+```php
+/** @var \EasySwoole\Queue\Queue $queue */
+$queue->consumer()->setOnBreak(function(\EasySwoole\Queue\Consumer $consumer) {
+     // todo 
+ })->listen(function (\EasySwoole\Queue\Job $job){ });
+```
+
+设置`breakTime`：
+
+```php
+/** @var \EasySwoole\Queue\Queue $queue */
+$queue->consumer()->setBreakTime(0.1);
+```

src/Consumer.php

@@ -9,7 +9,10 @@
 class Consumer
 {
     private $driver;
+
     private $enableListen = false;
+    private $onBreak;
+    private $breakTime = 0.01;
 
     function __construct(QueueDriverInterface $driver)
     {
@@ -31,7 +34,10 @@ function listen(callable $call,array $params = [],?callable $onException = null)
                 if ($job) {
                     call_user_func($call, $job);
                 } else {
-                    Coroutine::sleep(0.001);
+                    if($this->onBreak){
+                        call_user_func($this->onBreak,$this);
+                    }
+                    Coroutine::sleep($this->breakTime);
                 }
             }catch (\Throwable $throwable){
                 if($onException){
@@ -53,4 +59,16 @@ function stopListen(): Consumer
         $this->enableListen = false;
         return $this;
     }
+
+    function setOnBreak(callable $call): Consumer
+    {
+        $this->onBreak = $call;
+        return $this;
+    }
+
+    function setBreakTime(float $time): Consumer
+    {
+        $this->breakTime = $time;
+        return $this;
+    }
 }

src/Driver/RedisQueue.php

@@ -132,8 +132,19 @@ public function info(): ?array
             /** @var $redis Redis */
             return [
                 'runningQueue'=>$redis->lLen($this->queueName),
-                'delayQueue'=>$redis->zCard("{$this->queueName}_c")
+                'delayQueue'=>$redis->zCard("{$this->queueName}_d")
             ];
         });
     }
-}
+
+    public function flush():bool
+    {
+        $this->pool->invoke(function ($redis){
+            /** @var $redis Redis */
+            $redis->del("{$this->queueName}_c");
+            $redis->del("{$this->queueName}_d");
+            $redis->del("{$this->queueName}");
+        });
+        return true;
+    }
+}

src/QueueDriverInterface.php

@@ -13,4 +13,6 @@ public function pop(float $timeout = 3.0, array $params = []): ?Job;
     public function info(): ?array;
 
     public function confirm(Job $job,float $timeout = 3.0): bool;
+
+    public function flush():bool;
 }