Refactor TriggerSubscriber class to improve performance and add chann…

…el handling (#546)
friendsofhyperf · Feb 4, 2024 · 31a0aa8 · 31a0aa8
1 parent a8e12b1
commit 31a0aa8
Show file tree

Hide file tree

Showing 2 changed files with 66 additions and 5 deletions.
diff --git a/src/Subscriber/TriggerSubscriber.php b/src/Subscriber/TriggerSubscriber.php
@@ -11,10 +11,15 @@
 
 namespace FriendsOfHyperf\Trigger\Subscriber;
 
+use Closure;
 use FriendsOfHyperf\Trigger\Consumer;
 use FriendsOfHyperf\Trigger\Traits\Logger;
 use FriendsOfHyperf\Trigger\TriggerManager;
+use Hyperf\Coordinator\Constants;
+use Hyperf\Coordinator\CoordinatorManager;
 use Hyperf\Coroutine\Concurrent;
+use Hyperf\Engine\Channel;
+use Hyperf\Engine\Coroutine;
 use MySQLReplication\Definitions\ConstEventsNames;
 use MySQLReplication\Event\DTO\EventDTO;
 use MySQLReplication\Event\DTO\RowsDTO;
@@ -32,6 +37,10 @@ class TriggerSubscriber extends AbstractSubscriber
 
     protected ?LoggerInterface $logger = null;
 
+    protected ?Channel $chan = null;
+
+    protected int $channelSize = 65535;
+
     public function __construct(
         protected ContainerInterface $container,
         protected TriggerManager $triggerManager,
@@ -40,6 +49,9 @@ public function __construct(
         $this->concurrent = new Concurrent(
             (int) ($consumer->getOption('concurrent.limit') ?? 1)
         );
+        if ($consumer->getOption('channel.size')) {
+            $this->channelSize = (int) $consumer->getOption('channel.size');
+        }
         $this->logger = $this->getLogger();
     }
 
@@ -52,12 +64,63 @@ public static function getSubscribedEvents(): array
         ];
     }
 
+    protected function close(): void
+    {
+        $this->chan?->close();
+        $this->chan = null;
+    }
+
+    protected function loop(): void
+    {
+        if ($this->chan) {
+            return;
+        }
+
+        $this->chan = new Channel($this->channelSize);
+
+        Coroutine::create(function () {
+            try {
+                while (true) {
+                    while (true) {
+                        /** @var Closure|null $closure */
+                        $closure = $this->chan?->pop();
+
+                        if ($closure === null) {
+                            break 2;
+                        }
+
+                        try {
+                            $this->concurrent->create($closure);
+                        } catch (Throwable $e) {
+                            $this->error((string) $e);
+                            break;
+                        } finally {
+                            $closure = null;
+                        }
+                    }
+                }
+            } catch (Throwable $e) {
+                $this->error((string) $e);
+            } finally {
+                $this->close();
+            }
+        });
+
+        Coroutine::create(function () {
+            if (CoordinatorManager::until(Constants::WORKER_EXIT)->yield()) {
+                $this->close();
+            }
+        });
+    }
+
     protected function allEvents(EventDTO $event): void
     {
         if (! $event instanceof RowsDTO) {
             return;
         }
 
+        $this->loop();
+
         $key = join('.', [
             $this->consumer->getConnection(),
             $event->getTableMap()->getDatabase(),
@@ -69,7 +132,7 @@ protected function allEvents(EventDTO $event): void
 
         foreach ($this->triggerManager->get($key) as $callable) {
             foreach ($event->getValues() as $value) {
-                $this->concurrent->create(function () use ($callable, $value, $eventType) {
+                $this->chan->push(function () use ($callable, $value, $eventType) {
                     [$class, $method] = $callable;
 
                     if (! $this->container->has($class)) {

diff --git a/src/Traits/Logger.php b/src/Traits/Logger.php
@@ -11,7 +11,7 @@
 
 namespace FriendsOfHyperf\Trigger\Traits;
 
-use FriendsOfHyperf\Trigger\Contact\LoggerInterface as ContactLoggerInterface;
+use FriendsOfHyperf\Trigger\Contact\LoggerInterface as LoggerContact;
 use Hyperf\Context\ApplicationContext;
 use Hyperf\Contract\StdoutLoggerInterface;
 use Psr\Log\LoggerInterface;
@@ -46,7 +46,6 @@ protected function formatMessage(string $message, array $context = []): string
     {
         return sprintf(
             '[trigger%s] %s %s',
-            /* @phpstan-ignore-next-line */
             isset($this->connection) ? ".{$this->connection}" : 'default',
             $message,
             $context ? json_encode($context, JSON_UNESCAPED_UNICODE) : ''
@@ -55,15 +54,14 @@ protected function formatMessage(string $message, array $context = []): string
 
     protected function getLogger(): ?LoggerInterface
     {
-        /* @phpstan-ignore-next-line */
         if (isset($this->logger) && $this->logger instanceof LoggerInterface) {
             return $this->logger;
         }
 
         $container = ApplicationContext::getContainer();
 
         return match (true) {
-            $container->has(ContactLoggerInterface::class) => $container->get(ContactLoggerInterface::class),
+            $container->has(LoggerContact::class) => $container->get(LoggerContact::class),
             $container->has(StdoutLoggerInterface::class) => $container->get(StdoutLoggerInterface::class),
             default => null,
         };