优化rqueue消费的xgroup逻辑

workbunny · Jun 3, 2024 · 80df77f · 80df77f
1 parent 313a464
commit 80df77f
Showing 1 changed file with 9 additions and 3 deletions.
diff --git a/src/Builders/Traits/MessageQueueMethod.php b/src/Builders/Traits/MessageQueueMethod.php
@@ -15,6 +15,8 @@ trait MessageQueueMethod
 
     protected array $claimStartTags = [];
 
+    protected bool $_init = false;
+
     /**
      * @param string|null $queueName
      * @return array
@@ -392,9 +394,12 @@ public function consume(Worker $worker, bool $del = true): bool
             $consumerName = "$groupName-$worker->id";
             // create group
             $queueStreams = [];
-            foreach ($queues as $queueName) {
-                $client->xGroup('CREATE', $queueName, $groupName,'0', true);
-                $queueStreams[$queueName] = '>';
+            if (!$this->_init) {
+                foreach ($queues as $queueName) {
+                    $client->xGroup('CREATE', $queueName, $groupName,'0', true);
+                    $queueStreams[$queueName] = '>';
+                }
+                $this->_init = true;
             }
             // group read
             if($res = $client->xReadGroup(
@@ -473,6 +478,7 @@ public function consume(Worker $worker, bool $del = true): bool
         } catch (RedisException $exception) {
             Log::channel('plugin.workbunny.webman-rqueue.debug')?->debug($exception->getMessage(), $exception->getTrace());
             $this->getLogger()?->debug($exception->getMessage(), $exception->getTrace());
+            $this->_init = false;
             throw new WebmanRqueueException($exception->getMessage(), $exception->getCode(), $exception);
         }
     }