fix(instance): 优化实例心跳机制及IP获取逻辑

srako · srako · commit c60f13b4b380 · 2026-05-08T19:53:36.000+08:00
- 修复 Client 魔术方法对属性名的类型校验，提升异常准确性
- 新增 get_local_ip 函数，优先通过本地DNS解析获取非回环IP，失败则调用公网服务
- InstanceRegistrarProcess 支持获取本机IP作为实例IP，兼容容器和云环境
- 修改实例心跳逻辑，仅对临时实例发送心跳，默认 ephemeral 为 true
- 增加实例心跳失败计数，连续失败超过配置次数后重启进程，避免雪崩效应
- 兼容 Guzzle 请求参数映射，支持 FORM_PARAMS、JSON、BODY 等多种格式
- 多处 provider 请求改用异步事件循环版本，提升异步处理能力
- 配置新增实例心跳失败最大次数参数，默认值为3
- 更新配置监听器中 config_path 字段名称，统一配置文件路径标识
diff --git a/src/Client.php b/src/Client.php
@@ -129,7 +129,7 @@ public function getName(): ?string
      */
     public function __get($name)
     {
-        if (!isset($name) || !isset($this->alias[$name])) {
+        if (!is_string($name) || $name === '' || !isset($this->alias[$name])) {
             throw new NacosException("{$name} is invalid.");
         }
 
diff --git a/src/Process/AbstractProcess.php b/src/Process/AbstractProcess.php
@@ -46,7 +46,7 @@ protected function _stop(int|float $sleep = 0): void
         if ($sleep > 0) {
             Timer::add($sleep, function () {
                 Worker::stopAll();
-            });
+            }, [], false);
 
             return;
         }
diff --git a/src/Process/ConfigListenerProcess.php b/src/Process/ConfigListenerProcess.php
@@ -82,7 +82,10 @@ public function onWorkerStart(Worker $worker)
             Timer::add(0.0, (float) $this->longPullingInterval, function () {
                 $promises = [];
                 foreach ($this->configListeners as $listener) {
-                    list($dataId, $group, $tenant, $configPath) = $listener;
+                    $dataId = $listener['data_id'];
+                    $group = $listener['group_name'];
+                    $tenant = $listener['namespace_id'];
+                    $configPath = $listener['config_path'] ?? $listener['file_path'] ?? null;
                     if (file_exists($configPath)) {
                         $promises[] = $this->client->config->listenerAsync(
                             $dataId,
diff --git a/src/Process/InstanceRegistrarProcess.php b/src/Process/InstanceRegistrarProcess.php
@@ -7,6 +7,7 @@
 use GuzzleHttp\Exception\GuzzleException;
 use GuzzleHttp\Promise\Utils;
 use Psr\Http\Message\ResponseInterface;
+use function Workbunny\WebmanNacos\get_local_ip;
 use Workerman\Timer;
 use Workerman\Worker;
 
@@ -26,15 +27,28 @@ class InstanceRegistrarProcess extends AbstractProcess
      */
     protected array $heartbeatTimers = [];
 
+    /**
+     * 每个实例的连续心跳失败计数
+     * @var array<string,int>
+     */
+    protected array $heartbeatFailCount = [];
+
     /**
      * @var float
      */
     protected float $heartbeat;
 
+    /**
+     * 允许的最大连续心跳失败次数，超过才重启进程
+     * @var int
+     */
+    protected int $heartbeatFailMax;
+
     public function __construct()
     {
         parent::__construct();
         $this->heartbeat = (float) config('plugin.workbunny.webman-nacos.app.instance_heartbeat', 5.0);
+        $this->heartbeatFailMax = (int) config('plugin.workbunny.webman-nacos.app.instance_heartbeat_fail_max', 3);
     }
 
     /**
@@ -44,50 +58,97 @@ public function __construct()
      */
     protected function _heartbeat(string $name): void
     {
-        if (isset($this->instanceRegistrars[$name])) {
-            list($serviceName, $ip, $port, $option) = $this->instanceRegistrars[$name];
-            if (isset($option['ephemeral'])) {
-                $option['ephemeral'] = (is_string($option['ephemeral']) ? filter_var($option['ephemeral'], FILTER_VALIDATE_BOOLEAN, FILTER_NULL_ON_FAILURE) : (bool) $option['ephemeral']);
+        if (!isset($this->instanceRegistrars[$name])) {
+            return;
+        }
+        $instanceRegistrar = $this->instanceRegistrars[$name];
+        $serviceName = $instanceRegistrar['service_name'];
+        $ip = $instanceRegistrar['pod_ip'] ?: get_local_ip();
+        $port = $instanceRegistrar['pod_port'];
+        $option = $instanceRegistrar['options'] ?? [];
+
+        // 关键修复：Nacos OpenAPI 的 ephemeral 默认值为 true（临时实例）
+        // 当用户未显式指定时必须按 true 处理，否则会导致临时实例不发心跳而被 Nacos 摘除
+        if (array_key_exists('ephemeral', $option)) {
+            $ephemeral = is_string($option['ephemeral'])
+                ? filter_var($option['ephemeral'], FILTER_VALIDATE_BOOLEAN)
+                : (bool) $option['ephemeral'];
+        } else {
+            $ephemeral = true;
+        }
+        $option['ephemeral'] = $ephemeral;
+
+        // 仅对临时实例进行心跳（永久实例由服务端健康检查维持）
+        if (!$ephemeral) {
+            return;
+        }
+
+        $this->heartbeatFailCount[$name] = 0;
+        $this->heartbeatTimers[$name] = Timer::add($this->heartbeat, function () use ($name, $serviceName, $ip, $port, $option) {
+            if ($this->is_stopping) {
+                return;
             }
-            // 仅对非永久实例进行心跳
-            if ($option['ephemeral'] ?? false) {
-                $this->heartbeatTimers[$name] = Timer::add($this->heartbeat, function () use ($name, $serviceName, $ip, $port, $option) {
-                    if ($this->is_stopping) {
-                        return;
-                    }
-                    try {
-                        if (!$this->client->instance->beat(
-                            $serviceName,
-                            array_filter([
-                                    'ip'          => $ip,
-                                    'port'        => $port,
-                                    'serviceName' => $serviceName,
-                                ] + $option, fn ($value) => $value !== null),
-                            $option['groupName'] ?? null,
-                            $option['namespaceId'] ?? null,
-                            $option['ephemeral'] ?? null,
-                            false,
-                            $this->heartbeat
-                        )) {
-                            $this->logger()->error(
-                                "Nacos instance heartbeat failed: [0] {$this->client->instance->getMessage()}.",
-                                ['name' => $name, 'trace' => []]
-                            );
-                            $this->_stop($this->retry_interval);
+            // beat JSON 只保留 Nacos 识别的标准字段，避免非标字段污染
+            $beatData = array_filter([
+                'ip'          => $ip,
+                'port'        => $port,
+                'serviceName' => $serviceName,
+                'cluster'     => $option['clusterName'] ?? ($option['cluster'] ?? null),
+                'weight'      => $option['weight'] ?? null,
+                'metadata'    => $option['metadata'] ?? null,
+                'scheduled'   => $option['scheduled'] ?? null,
+            ], fn ($value) => $value !== null);
 
-                            return;
-                        }
-                    } catch (GuzzleException $exception) {
-                        $this->logger()->error(
-                            "Nacos instance heartbeat failed: [{$exception->getCode()}] {$exception->getMessage()}.",
-                            ['name' => $name, 'trace' => $exception->getTrace()]
-                        );
-                        $this->_stop($this->retry_interval);
-
-                        return;
-                    }
-                });
+            try {
+                $result = $this->client->instance->beat(
+                    $serviceName,
+                    $beatData,
+                    $option['groupName'] ?? null,
+                    $option['namespaceId'] ?? null,
+                    $option['ephemeral'] ?? null,
+                    false,
+                    $this->heartbeat
+                );
+                if ($result === false) {
+                    $this->_onHeartbeatFail(
+                        $name,
+                        "Nacos instance heartbeat failed: [0] {$this->client->instance->getMessage()}.",
+                        []
+                    );
+
+                    return;
+                }
+                // 心跳成功，重置失败计数
+                $this->heartbeatFailCount[$name] = 0;
+            } catch (\Throwable $exception) {
+                $this->_onHeartbeatFail(
+                    $name,
+                    "Nacos instance heartbeat failed: [{$exception->getCode()}] {$exception->getMessage()}.",
+                    $exception->getTrace()
+                );
             }
+        });
+    }
+
+    /**
+     * 心跳失败处理：累计 N 次失败才重启进程，避免偶发抖动导致雪崩
+     * @param string $name
+     * @param string $message
+     * @param array $trace
+     * @return void
+     */
+    protected function _onHeartbeatFail(string $name, string $message, array $trace = []): void
+    {
+        $this->heartbeatFailCount[$name] = ($this->heartbeatFailCount[$name] ?? 0) + 1;
+        $count = $this->heartbeatFailCount[$name];
+        $this->logger()->error($message, [
+            'name'      => $name,
+            'fail_count'=> $count,
+            'max'       => $this->heartbeatFailMax,
+            'trace'     => $trace,
+        ]);
+        if ($count >= $this->heartbeatFailMax) {
+            $this->_stop($this->retry_interval);
         }
     }
 
@@ -103,7 +164,7 @@ public function onWorkerStart(Worker $worker)
                 foreach ($instanceRegistrars as $name => $instanceRegistrar) {
                     // 拆解配置
                     $serviceName = $instanceRegistrar['service_name'];
-                    $ip = $instanceRegistrar['pod_ip'];
+                    $ip = $instanceRegistrar['pod_ip'] ?: get_local_ip();
                     $port = $instanceRegistrar['pod_port'];
                     $option = $instanceRegistrar['options'] ?? [];
                     // 注册
@@ -153,18 +214,19 @@ public function onWorkerStop(Worker $worker)
                 }
                 // 拆解配置
                 $serviceName = $instanceRegistrar['service_name'];
-                $ip = $instanceRegistrar['pod_ip'];
+                $ip = $instanceRegistrar['pod_ip'] ?: get_local_ip();
                 $port = $instanceRegistrar['pod_port'];
                 $option = $instanceRegistrar['options'] ?? [];
-                // 注销实例
+                // 注销实例（groupName 未配置时使用 Nacos 默认的 DEFAULT_GROUP，避免 null 触发 TypeError）
                 if (!$this->client->instance->delete(
                     $serviceName,
-                    $option['groupName'] ?? null,
+                    (string) ($option['groupName'] ?? 'DEFAULT_GROUP'),
                     $ip,
                     $port,
                     [
                         'namespaceId' => $option['namespaceId'] ?? null,
                         'ephemeral'   => $option['ephemeral'] ?? null,
+                        'clusterName' => $option['clusterName'] ?? null,
                     ]
                 )) {
                     $this->logger()->error(
@@ -173,7 +235,7 @@ public function onWorkerStop(Worker $worker)
                     );
                 }
             }
-        } catch (GuzzleException $exception) {
+        } catch (\Throwable $exception) {
             $this->logger()->error(
                 "Nacos instance delete failed: [{$exception->getCode()}] {$exception->getMessage()}.",
                 ['name' => '#base', 'trace' => $exception->getTrace()]
diff --git a/src/Provider/AbstractProvider.php b/src/Provider/AbstractProvider.php
@@ -217,13 +217,25 @@ public function requestAsyncUseEventLoop(string $method, string $uri, array $opt
             $headers = array_merge($options[RequestOptions::HEADERS] ?? [], [
                 'Connection' => 'keep-alive',
             ]);
+            // 兼容 Guzzle 的 FORM_PARAMS / JSON / BODY，映射到 workerman/http-client 的 data
+            $data = [];
+            if (isset($options[RequestOptions::FORM_PARAMS])) {
+                $data = $options[RequestOptions::FORM_PARAMS];
+            } elseif (isset($options[RequestOptions::JSON])) {
+                $data = json_encode($options[RequestOptions::JSON]);
+                $headers['Content-Type'] = 'application/json';
+            } elseif (isset($options[RequestOptions::BODY])) {
+                $data = $options[RequestOptions::BODY];
+            } elseif (isset($options['data'])) {
+                $data = $options['data'];
+            }
             $this->httpClientAsync()->request(
                 sprintf('http://%s:%d/%s?%s', $this->host, $this->port, $uri, $queryString),
                 [
                     'method'    => $method,
                     'version'   => '1.1',
                     'headers'   => $headers,
-                    'data'      => $options['data'] ?? [],
+                    'data'      => $data,
                     'success'   => $options['success'] ?? function (Response $response) {
                     },
                     'error'     => $options['error'] ?? function (\Exception $exception) {
diff --git a/src/Provider/ConfigProvider.php b/src/Provider/ConfigProvider.php
@@ -329,6 +329,7 @@ public function listenerAsyncUseEventLoop(array $options, ?callable $success = n
             ($options['group'] ?? null) . self::WORD_SEPARATOR .
             ($options['contentMD5'] ?? null) . self::WORD_SEPARATOR .
             ($options['tenant'] ?? null) . self::LINE_SEPARATOR;
+        $timeout = $options['timeout'] ?? null;
 
         return $this->requestAsyncUseEventLoop(self::LISTENER_METHOD, self::LISTENER_URL, [
             RequestOptions::QUERY   => [
diff --git a/src/Provider/InstanceProvider.php b/src/Provider/InstanceProvider.php
@@ -420,7 +420,7 @@ public function detail(string $ip, int $port, string $serviceName, array $option
      */
     public function detailAsync(string $ip, int $port, string $serviceName, array $optional = [])
     {
-        return $this->request(self::DETAIL_METHOD, self::DETAIL_URL, [
+        return $this->requestAsync(self::DETAIL_METHOD, self::DETAIL_URL, [
             RequestOptions::QUERY => $this->filter(array_merge($optional, [
                 'ip'          => $ip,
                 'port'        => $port,
@@ -665,7 +665,7 @@ public function beatAsyncUseEventLoop(array $options, bool $lightBeatEnabled = f
             ['timeout', 'is_float', false],
         ]);
 
-        return $this->requestAsync(self::BEAT_METHOD, self::BEAT_URL, [
+        return $this->requestAsyncUseEventLoop(self::BEAT_METHOD, self::BEAT_URL, [
             RequestOptions::QUERY => $this->filter([
                 'serviceName' => $options['serviceName'] ?? null,
                 'ip'          => $options['beat']['ip'] ?? null,
diff --git a/src/Provider/OperatorProvider.php b/src/Provider/OperatorProvider.php
@@ -237,7 +237,7 @@ public function getLeaderAsync()
      */
     public function getLeaderAsyncUseEventLoop(?callable $success = null, ?callable $error = null): bool
     {
-        return $this->requestAsync(self::GET_LEADER_METHOD, self::GET_LEADER_URL, [
+        return $this->requestAsyncUseEventLoop(self::GET_LEADER_METHOD, self::GET_LEADER_URL, [
             OPTIONS_SUCCESS => $success,
             OPTIONS_ERROR   => $error,
         ]);
diff --git a/src/Provider/ServiceProvider.php b/src/Provider/ServiceProvider.php
@@ -157,7 +157,7 @@ public function deleteAsyncUseEventLoop(array $options, ?callable $success = nul
             ['namespaceId', 'is_string', false],
         ]);
 
-        return $this->requestAsync(self::DELETE_METHOD, self::DELETE_URL, [
+        return $this->requestAsyncUseEventLoop(self::DELETE_METHOD, self::DELETE_URL, [
             RequestOptions::QUERY => $this->filter([
                 'serviceName' => $options['serviceName'] ?? null,
                 'groupName'   => $options['groupName'] ?? null,
@@ -226,7 +226,7 @@ public function updateAsync(string $serviceName, array $optional = [])
      */
     public function updateAsyncUseEventLoop(string $serviceName, array $optional = [], ?callable $success = null, ?callable $error = null): bool
     {
-        return $this->requestAsync(self::UPDATE_METHOD, self::UPDATE_URL, [
+        return $this->requestAsyncUseEventLoop(self::UPDATE_METHOD, self::UPDATE_URL, [
             RequestOptions::QUERY => $this->filter(array_merge($optional, [
                 'serviceName' => $serviceName,
             ])),
@@ -290,7 +290,7 @@ public function getAsyncUseEventLoop(array $options, ?callable $success = null,
             ['namespaceId', 'is_string', false],
         ]);
 
-        return $this->requestAsync(self::GET_METHOD, self::GET_URL, [
+        return $this->requestAsyncUseEventLoop(self::GET_METHOD, self::GET_URL, [
             RequestOptions::QUERY => $this->filter([
                 'serviceName' => $options['serviceName'] ?? null,
                 'groupName'   => $options['groupName'] ?? null,
@@ -362,7 +362,7 @@ public function listAsyncUseEventLoop(array $options, ?callable $success = null,
             ['namespaceId', 'is_string', false],
         ]);
 
-        return $this->requestAsync(self::LIST_METHOD, self::LIST_URL, [
+        return $this->requestAsyncUseEventLoop(self::LIST_METHOD, self::LIST_URL, [
             RequestOptions::QUERY => $this->filter([
                 'pageNo'      => $options['pageNo'] ?? null,
                 'pageSize'    => $options['pageSize'] ?? null,
diff --git a/src/config/plugin/workbunny/webman-nacos/app.php b/src/config/plugin/workbunny/webman-nacos/app.php
@@ -10,6 +10,8 @@
     'long_pulling_interval'  => 30,
     /* float 实例心跳间隔 秒 */
     'instance_heartbeat' => 5.0,
+    /* int 实例心跳连续失败达该次数后才重启进程，避免偶发抖动导致服务摘除 */
+    'instance_heartbeat_fail_max' => 3,
     /* int 进程重试间隔 秒 */
     'process_retry_interval' => 5,
     /* 日志 channel，为 null 时使用默认通道 */
diff --git a/src/config/plugin/workbunny/webman-nacos/listeners.php b/src/config/plugin/workbunny/webman-nacos/listeners.php
@@ -7,6 +7,6 @@
         'data_id'      => 'config.yaml',
         'group_name'   => '{your_group_name}',
         'namespace_id' => '{your_namespace_id}',
-        'file_path'    => base_path() . '/config.yaml',
+        'config_path'    => base_path() . '/config.yaml',
     ],
 ];
diff --git a/src/helpers.php b/src/helpers.php
@@ -38,11 +38,35 @@ function reload(string $file): bool
 /**
  * 获取本机ip
  *
- * @return string
+ * 优先级：本地DNS解析（过滤回环） > 外部服务探测
+ *
+ * @return string 非回环IP，获取失败返回空字符串
  */
 function get_local_ip(): string
 {
-    return trim(shell_exec('curl -s ifconfig.me'));
+    // 1. 通过本地DNS解析获取
+    $ips = gethostbynamel(gethostname());
+    if ($ips !== false) {
+        // 过滤回环地址，优先 IPv4
+        $ips = array_values(array_filter($ips, function (string $ip): bool {
+            return $ip !== '127.0.0.1' && $ip !== '::1';
+        }));
+        if ($ips) {
+            // 多网卡时优先取私有地址（10.x / 172.16-31.x / 192.168.x），更可能是实际通信IP
+            $private = array_filter($ips, function (string $ip): bool {
+                return (bool) preg_match('/^(10\.|172\.(1[6-9]|2\d|3[01])\.|192\.168\.)/', $ip);
+            });
+            return $private ? array_pop($private) : array_pop($ips);
+        }
+    }
+
+    // 2. fallback: 通过外部服务获取公网IP（适用于容器/云主机 where hostname 无法解析出内网IP）
+    $ip = trim((string) shell_exec('curl -s --connect-timeout 3 --max-time 5 ifconfig.me 2>/dev/null'));
+    if ($ip !== '' && filter_var($ip, FILTER_VALIDATE_IP)) {
+        return $ip;
+    }
+
+    return '';
 }
 
 /**

Original file line number	Diff line number	Diff line change
`@@ -129,7 +129,7 @@ public function getName(): ?string`
`129`	`129`	`*/`
`130`	`130`	`public function __get($name)`
`131`	`131`	`{`
`132`		`- if (!isset($name) \|\| !isset($this->alias[$name])) {`
	`132`	`+ if (!is_string($name) \|\| $name === '' \|\| !isset($this->alias[$name])) {`
`133`	`133`	`throw new NacosException("{$name} is invalid.");`
`134`	`134`	`}`
`135`	`135`
Original file line number	Diff line number	Diff line change
`@@ -46,7 +46,7 @@ protected function _stop(int\|float $sleep = 0): void`
`46`	`46`	`if ($sleep > 0) {`
`47`	`47`	`Timer::add($sleep, function () {`
`48`	`48`	`Worker::stopAll();`
`49`		`- });`
	`49`	`+ }, [], false);`
`50`	`50`
`51`	`51`	`return;`
`52`	`52`	`}`
Original file line number	Diff line number	Diff line change
`@@ -237,7 +237,7 @@ public function getLeaderAsync()`
`237`	`237`	`*/`
`238`	`238`	`public function getLeaderAsyncUseEventLoop(?callable $success = null, ?callable $error = null): bool`
`239`	`239`	`{`
`240`		`- return $this->requestAsync(self::GET_LEADER_METHOD, self::GET_LEADER_URL, [`
	`240`	`+ return $this->requestAsyncUseEventLoop(self::GET_LEADER_METHOD, self::GET_LEADER_URL, [`
`241`	`241`	`OPTIONS_SUCCESS => $success,`
`242`	`242`	`OPTIONS_ERROR => $error,`
`243`	`243`	`]);`