Optimized exchange declaration for consumer messages (#6716)

hyperf · Apr 26, 2024 · 0d6e7ac · 0d6e7ac
1 parent 2436f4f
commit 0d6e7ac
Showing 1 changed file with 11 additions and 2 deletions.
diff --git a/src/Consumer.php b/src/Consumer.php
@@ -55,10 +55,19 @@ public function consume(ConsumerMessageInterface $consumerMessage): void
 
         try {
             $channel = $connection->getConfirmChannel();
+            $exchange = $consumerMessage->getExchange();
 
-            $this->declare($consumerMessage, $channel);
-            $concurrent = $this->getConcurrent($consumerMessage->getPoolName());
+            if (! DeclaredExchanges::has($exchange)) {
+                try {
+                    DeclaredExchanges::add($exchange);
+                    $this->declare($consumerMessage, $channel);
+                } catch (Throwable $exception) {
+                    DeclaredExchanges::remove($exchange);
+                    throw $exception;
+                }
+            }
 
+            $concurrent = $this->getConcurrent($consumerMessage->getPoolName());
             $maxConsumption = $consumerMessage->getMaxConsumption();
             $currentConsumption = 0;