Merge pull request #1419 from Reasno/config

Allow config fetcher to start in a coroutine instead of a process

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## Added
 
 - [#1393](https://github.com/hyperf/hyperf/pull/1393) Implements methods for `Hyperf\HttpMessage\Stream\SwooleStream`.
+- [#1419](https://github.com/hyperf/hyperf/pull/1419) Allow config fetcher to start in a coroutine instead of a process.
 - [#1424](https://github.com/hyperf/hyperf/pull/1424) Allow user modify the session_name by configuration file.
 
 # v1.1.20 - 2020-03-12

doc/zh-cn/config-center.md

@@ -40,6 +40,8 @@ composer require hyperf/config-aliyun-acm
 return [
     // 是否开启配置中心的接入流程，为 true 时会自动启动一个 ConfigFetcherProcess 进程用于更新配置
     'enable' => true,
+    // 是否使用独立进程来拉取config，如果否则将在worker内以协程方式拉取
+    'use_standalone_process' => true,
     // Apollo Server
     'server' => 'http://127.0.0.1:8080',
     // 您的 AppId
@@ -74,6 +76,8 @@ return [
 return [
     // 是否开启配置中心的接入流程，为 true 时会自动启动一个 ConfigFetcherProcess 进程用于更新配置
     'enable' => true,
+    // 是否使用独立进程来拉取config，如果否则将在worker内以协程方式拉取
+    'use_standalone_process' => true,
     // 配置更新间隔（秒）
     'interval' => 5,
     // 阿里云 ACM 断点地址，取决于您的可用区
@@ -135,6 +139,7 @@ composer require hyperf/config-etcd
 <?php
 return [
     'enable' => true,
+    'use_standalone_process' => true,
     'namespaces' => [
         '/test',
     ],

src/config-aliyun-acm/publish/aliyun_acm.php

@@ -12,6 +12,7 @@ declare(strict_types=1);
 
 return [
     'enable' => false,
+    'use_standalone_process' => true,
     'interval' => 5,
     'endpoint' => env('ALIYUN_ACM_ENDPOINT', 'acm.aliyun.com'),
     'namespace' => env('ALIYUN_ACM_NAMESPACE', ''),

src/config-aliyun-acm/src/Listener/BootProcessListener.php

@@ -12,12 +12,14 @@ declare(strict_types=1);
 
 namespace Hyperf\ConfigAliyunAcm\Listener;
 
+use Hyperf\Command\Event\BeforeHandle;
 use Hyperf\ConfigAliyunAcm\ClientInterface;
 use Hyperf\Contract\ConfigInterface;
 use Hyperf\Contract\StdoutLoggerInterface;
 use Hyperf\Event\Contract\ListenerInterface;
 use Hyperf\Framework\Event\BeforeWorkerStart;
 use Hyperf\Process\Event\BeforeProcessHandle;
+use Hyperf\Utils\Coroutine;
 use Psr\Container\ContainerInterface;
 
 class BootProcessListener implements ListenerInterface
@@ -49,6 +51,7 @@ class BootProcessListener implements ListenerInterface
         return [
             BeforeWorkerStart::class,
             BeforeProcessHandle::class,
+            BeforeHandle::class,
         ];
     }
 
@@ -59,11 +62,33 @@ class BootProcessListener implements ListenerInterface
         }
 
         if ($config = $this->client->pull()) {
-            foreach ($config as $key => $value) {
+            $this->updateConfig($config);
-                if (is_string($key)) {
+        }
-                    $this->config->set($key, $value);
+
-                    $this->logger->debug(sprintf('Config [%s] is updated', $key));
+        if (! $this->config->get('aliyun_acm.use_standalone_process', true)) {
-                }
+            Coroutine::create(function () {
+                $interval = $this->config->get('aliyun_acm.interval', 5);
+                retry(INF, function () use ($interval) {
+                    $prevConfig = [];
+                    while (true) {
+                        sleep($interval);
+                        $config = $this->client->pull();
+                        if ($config !== $prevConfig) {
+                            $this->updateConfig($config);
+                        }
+                        $prevConfig = $config;
+                    }
+                }, $interval * 1000);
+            });
+        }
+    }
+
+    protected function updateConfig(array $config)
+    {
+        foreach ($config as $key => $value) {
+            if (is_string($key)) {
+                $this->config->set($key, $value);
+                $this->logger->debug(sprintf('Config [%s] is updated', $key));
             }
         }
     }

src/config-aliyun-acm/src/Process/ConfigFetcherProcess.php

@@ -59,7 +59,8 @@ class ConfigFetcherProcess extends AbstractProcess
 
     public function isEnable(): bool
     {
-        return $this->config->get('aliyun_acm.enable', false);
+        return $this->config->get('aliyun_acm.enable', false)
+            && $this->config->get('aliyun_acm.use_standalone_process', true);
     }
 
     public function handle(): void

src/config-apollo/publish/apollo.php

@@ -12,6 +12,7 @@ declare(strict_types=1);
 
 return [
     'enable' => false,
+    'use_standalone_process' => true,
     'server' => 'http://localhost:8080',
     'appid' => 'test',
     'cluster' => 'default',

src/config-apollo/src/Listener/BootProcessListener.php

@@ -12,11 +12,13 @@ declare(strict_types=1);
 
 namespace Hyperf\ConfigApollo\Listener;
 
+use Hyperf\Command\Event\BeforeHandle;
 use Hyperf\ConfigApollo\Option;
 use Hyperf\ConfigApollo\PipeMessage;
 use Hyperf\ConfigApollo\ReleaseKey;
 use Hyperf\Framework\Event\BeforeWorkerStart;
 use Hyperf\Process\Event\BeforeProcessHandle;
+use Hyperf\Utils\Coroutine;
 
 class BootProcessListener extends OnPipeMessageListener
 {
@@ -25,6 +27,7 @@ class BootProcessListener extends OnPipeMessageListener
         return [
             BeforeWorkerStart::class,
             BeforeProcessHandle::class,
+            BeforeHandle::class,
         ];
     }
 
@@ -62,5 +65,17 @@ class BootProcessListener extends OnPipeMessageListener
             $callbacks[$namespace] = $ipcCallback;
         }
         $this->client->pull($namespaces, $callbacks);
+
+        if (! $this->config->get('apollo.use_standalone_process', true)) {
+            Coroutine::create(function () use ($namespaces, $callbacks) {
+                $interval = $this->config->get('apollo.interval', 5);
+                retry(INF, function () use ($namespaces, $callbacks, $interval) {
+                    while (true) {
+                        sleep($interval);
+                        $this->client->pull($namespaces, $callbacks);
+                    }
+                }, $interval * 1000);
+            });
+        }
     }
 }

src/config-apollo/src/Process/ConfigFetcherProcess.php

@@ -54,7 +54,8 @@ class ConfigFetcherProcess extends AbstractProcess
 
     public function isEnable(): bool
     {
-        return $this->config->get('apollo.enable', false);
+        return $this->config->get('apollo.enable', false)
+            && $this->config->get('apollo.use_standalone_process', true);
     }
 
     public function handle(): void

src/config-etcd/publish/config_etcd.php

@@ -13,6 +13,7 @@ declare(strict_types=1);
 return [
     'enable' => false,
     'packer' => Hyperf\Utils\Packer\JsonPacker::class,
+    'use_standalone_process' => true,
     'namespaces' => [
         'application',
     ],

src/config-etcd/src/Listener/BootProcessListener.php

@@ -12,6 +12,7 @@ declare(strict_types=1);
 
 namespace Hyperf\ConfigEtcd\Listener;
 
+use Hyperf\Command\Event\BeforeHandle;
 use Hyperf\ConfigEtcd\ClientInterface;
 use Hyperf\ConfigEtcd\KV;
 use Hyperf\Contract\ConfigInterface;
@@ -20,6 +21,7 @@ use Hyperf\Contract\StdoutLoggerInterface;
 use Hyperf\Event\Contract\ListenerInterface;
 use Hyperf\Framework\Event\BeforeWorkerStart;
 use Hyperf\Process\Event\BeforeProcessHandle;
+use Hyperf\Utils\Coroutine;
 use Hyperf\Utils\Packer\JsonPacker;
 use Psr\Container\ContainerInterface;
 
@@ -65,6 +67,7 @@ class BootProcessListener implements ListenerInterface
         return [
             BeforeWorkerStart::class,
             BeforeProcessHandle::class,
+            BeforeHandle::class,
         ];
     }
 
@@ -75,13 +78,35 @@ class BootProcessListener implements ListenerInterface
         }
 
         if ($config = $this->client->pull()) {
-            $configurations = $this->format($config);
+            $this->updateConfig($config);
-            foreach ($configurations as $kv) {
+        }
-                $key = $this->mapping[$kv->key] ?? null;
+
-                if (is_string($key)) {
+        if (! $this->config->get('config_etcd.use_standalone_process', true)) {
-                    $this->config->set($key, $this->packer->unpack($kv->value));
+            Coroutine::create(function () {
-                    $this->logger->debug(sprintf('Config [%s] is updated', $key));
+                $interval = $this->config->get('config_etcd.interval', 5);
-                }
+                retry(INF, function () use ($interval) {
+                    $prevConfig = [];
+                    while (true) {
+                        sleep($interval);
+                        $config = $this->client->pull();
+                        if ($config !== $prevConfig) {
+                            $this->updateConfig($config);
+                        }
+                        $prevConfig = $config;
+                    }
+                }, $interval * 1000);
+            });
+        }
+    }
+
+    protected function updateConfig(array $config)
+    {
+        $configurations = $this->format($config);
+        foreach ($configurations as $kv) {
+            $key = $this->mapping[$kv->key] ?? null;
+            if (is_string($key)) {
+                $this->config->set($key, $this->packer->unpack($kv->value));
+                $this->logger->debug(sprintf('Config [%s] is updated', $key));
             }
         }
     }

src/config-etcd/src/Process/ConfigFetcherProcess.php

@@ -60,7 +60,8 @@ class ConfigFetcherProcess extends AbstractProcess
 
     public function isEnable(): bool
     {
-        return $this->config->get('config_etcd.enable', false);
+        return $this->config->get('config_etcd.enable', false)
+            && $this->config->get('config_etcd.use_standalone_process', true);
     }
 
     public function handle(): void

src/config-zookeeper/publish/zookeeper.php

@@ -12,6 +12,7 @@ declare(strict_types=1);
 
 return [
     'enable' => false,
+    'use_standalone_process' => true,
     'interval' => 5,
     'server' => env('ZOOKEEPER_SERVER', '127.0.0.1:2181'),
     'path' => env('ZOOKEEPER_CONFIG_PATH', '/conf'),

src/config-zookeeper/src/ConfigProvider.php

@@ -12,6 +12,7 @@ declare(strict_types=1);
 
 namespace Hyperf\ConfigZookeeper;
 
+use Hyperf\ConfigZookeeper\Listener\BootProcessListener;
 use Hyperf\ConfigZookeeper\Listener\OnPipeMessageListener;
 use Hyperf\ConfigZookeeper\Process\ConfigFetcherProcess;
 
@@ -28,6 +29,7 @@ class ConfigProvider
             ],
             'listeners' => [
                 OnPipeMessageListener::class,
+                BootProcessListener::class,
             ],
             'annotations' => [
                 'scan' => [

src/config-zookeeper/src/Listener/BootProcessListener.php

@@ -0,0 +1,90 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://www.hyperf.io
+ * @document https://doc.hyperf.io
+ * @contact  group@hyperf.io
+ * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+
+namespace Hyperf\ConfigZookeeper\Listener;
+
+use Hyperf\Command\Event\BeforeHandle;
+use Hyperf\ConfigZookeeper\ClientInterface;
+use Hyperf\Contract\ConfigInterface;
+use Hyperf\Contract\StdoutLoggerInterface;
+use Hyperf\Event\Contract\ListenerInterface;
+use Hyperf\Framework\Event\BeforeWorkerStart;
+use Hyperf\Process\Event\BeforeProcessHandle;
+use Hyperf\Utils\Coroutine;
+
+class BootProcessListener implements ListenerInterface
+{
+    /**
+     * @var ConfigInterface
+     */
+    private $config;
+
+    /**
+     * @var StdoutLoggerInterface
+     */
+    private $logger;
+
+    /**
+     * @var ClientInterface
+     */
+    private $client;
+
+    public function __construct(ConfigInterface $config, StdoutLoggerInterface $logger, ClientInterface $client)
+    {
+        $this->config = $config;
+        $this->logger = $logger;
+        $this->client = $client;
+    }
+
+    public function listen(): array
+    {
+        return [
+            BeforeWorkerStart::class,
+            BeforeProcessHandle::class,
+            BeforeHandle::class,
+        ];
+    }
+
+    public function process(object $event)
+    {
+        if (! $this->config->get('zookeeper.enable', false)) {
+            return;
+        }
+
+        if (! $this->config->get('zookeeper.use_standalone_process', true)) {
+            Coroutine::create(function () {
+                $interval = $this->config->get('zookeeper.interval', 5);
+                retry(INF, function () use ($interval) {
+                    $prevConfig = [];
+                    while (true) {
+                        sleep($interval);
+                        $config = $this->client->pull();
+                        if ($config !== $prevConfig) {
+                            $this->updateConfig($config);
+                        }
+                        $prevConfig = $config;
+                    }
+                }, $interval * 1000);
+            });
+        }
+    }
+
+    protected function updateConfig(array $config)
+    {
+        foreach ($config as $key => $value) {
+            if (is_string($key)) {
+                $this->config->set($key, $value);
+                $this->logger->debug(sprintf('Config [%s] is updated', $key));
+            }
+        }
+    }
+}

src/config-zookeeper/src/Process/ConfigFetcherProcess.php

@@ -63,7 +63,8 @@ class ConfigFetcherProcess extends AbstractProcess
 
     public function isEnable(): bool
     {
-        return $this->config->get('zookeeper.enable', false);
+        return $this->config->get('zookeeper.enable', false)
+            && $this->config->get('zookeeper.use_standalone_process', true);
     }
 
     public function handle(): void