[fit] 代码重构

Author: mikeaaaaaa

framework/fit/python/conf/application.yml

@@ -42,7 +42,7 @@ https:
     key_file_password_encrypted: false # 私钥的密码是否被加密，仅当 cert_enabled 和 key_file_encrypted 为 true 时有意义
 registry-center:
   server:
-    mode: 'DIRECT'
+    mode: 'DIRECT' # DIRECT 表示直连，直接连接内存注册中心；PROXY 表示代理模式，通过本地代理服务连接 Nacos 注册中心
     addresses:
       - "localhost:8848"
     protocol: 2

framework/fit/python/plugin/fit_py_nacos_registry/service/__init__.py

@@ -0,0 +1,26 @@
+# -- encoding: utf-8 --
+# Copyright (c) 2025 Huawei Technologies Co., Ltd. All Rights Reserved.
+# This file is a part of the ModelEngine Project.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# ======================================================================================================================
+"""
+Nacos registry service package.
+"""
+
+from .nacos_registry_server import (
+    register_fitables,
+    unregister_fitables,
+    query_fitable_addresses,
+    subscribe_fit_service,
+    unsubscribe_fitables,
+    query_fitable_metas
+)
+
+__all__ = [
+    'register_fitables',
+    'unregister_fitables',
+    'query_fitable_addresses',
+    'subscribe_fit_service',
+    'unsubscribe_fitables',
+    'query_fitable_metas'
+]

framework/fit/python/plugin/fit_py_nacos_registry/service/async_executor.py

@@ -0,0 +1,212 @@
+# -- encoding: utf-8 --
+# Copyright (c) 2025 Huawei Technologies Co., Ltd. All Rights Reserved.
+# This file is a part of the ModelEngine Project.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# ======================================================================================================================
+"""
+Async executor for Nacos operations.
+
+This module provides an async executor for handling Nacos operations
+in a background thread with proper event loop management.
+"""
+import asyncio
+import atexit
+import threading
+from concurrent.futures import Future
+
+from v2.nacos import NacosNamingService, RegisterInstanceParam, ListInstanceParam, \
+    DeregisterInstanceParam, SubscribeServiceParam, ListServiceParam
+
+from fitframework.api.logging import plugin_logger
+from .config import build_nacos_config
+
+
+class AsyncExecutor:
+    """Executor for handling asynchronous operations in a background thread."""
+
+    def __init__(self):
+        self._loop = None
+        self._thread = None
+        self._started = False
+        self._shutdown = False
+        self._nacos_client = None
+        self._init_complete = threading.Event()
+
+    def start(self):
+        """Start the background event loop thread."""
+        if self._started:
+            return
+
+        self._thread = threading.Thread(
+            target=self._run_event_loop, 
+            daemon=True, 
+            name="NacosAsyncThread"
+        )
+        self._thread.start()
+
+        # Wait for initialization to complete
+        if not self._init_complete.wait(timeout=10):  # Max wait 10 seconds
+            raise RuntimeError("Failed to initialize async executor within timeout")
+
+        self._started = True
+
+    def _run_event_loop(self):
+        """Run the event loop in the background thread."""
+        try:
+            self._loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(self._loop)
+
+            # Create Nacos client in this event loop
+            async def init_nacos_client():
+                try:
+                    config = build_nacos_config()
+                    self._nacos_client = await NacosNamingService.create_naming_service(config)
+                    plugin_logger.info("Nacos client initialized successfully")
+                except Exception as e:
+                    plugin_logger.error(f"Failed to initialize Nacos client: {e}")
+                    raise
+                finally:
+                    # Mark initialization complete
+                    self._init_complete.set()
+
+            self._loop.run_until_complete(init_nacos_client())
+
+            # Run event loop until shutdown
+            self._loop.run_forever()
+        except Exception as e:
+            plugin_logger.error(f"Error in async executor event loop: {e}")
+            self._init_complete.set()  # Set even on failure to avoid infinite wait
+        finally:
+            try:
+                if self._nacos_client:
+                    # Cleanup Nacos client if needed
+                    pass
+                if self._loop:
+                    self._loop.close()
+            except Exception as e:
+                plugin_logger.error(f"Error during cleanup: {e}")
+
+    def run_coroutine(self, coro):
+        """
+        Run a coroutine in the background event loop and return the result.
+        
+        Args:
+            coro: The coroutine to run.
+            
+        Returns:
+            The result of the coroutine.
+            
+        Raises:
+            RuntimeError: If the executor is not properly initialized.
+        """
+        if not self._started:
+            self.start()
+
+        if self._loop is None or self._nacos_client is None:
+            raise RuntimeError("Async executor not properly initialized")
+
+        # Create a Future to get the result
+        result_future = Future()
+
+        async def wrapped_coro():
+            try:
+                result = await coro
+                result_future.set_result(result)
+            except Exception as e:
+                result_future.set_exception(e)
+
+        # Schedule the coroutine in the event loop
+        self._loop.call_soon_threadsafe(asyncio.create_task, wrapped_coro())
+
+        # Wait for result
+        return result_future.result(timeout=30)  # 30 second timeout
+
+    def get_nacos_client(self):
+        """
+        Get the Nacos client instance.
+        
+        Returns:
+            The Nacos client instance.
+        """
+        if not self._started:
+            self.start()
+        return self._nacos_client
+
+    def shutdown(self):
+        """Shutdown the async executor."""
+        if self._loop and not self._loop.is_closed():
+            self._loop.call_soon_threadsafe(self._loop.stop)
+        self._shutdown = True
+
+
+# Global async executor
+_async_executor = AsyncExecutor()
+
+
+def run_async_safely(coro):
+    """
+    Run an async operation safely using the dedicated executor.
+
+    Args:
+        coro: The coroutine to run.
+
+    Returns:
+        The result of the coroutine.
+        
+    Raises:
+        Exception: If the async operation fails.
+    """
+    try:
+        return _async_executor.run_coroutine(coro)
+    except Exception as e:
+        plugin_logger.error(f"Error running async operation: {e}")
+        raise
+
+
+# Async wrapper functions
+async def call_list_instances(param: ListInstanceParam):
+    """List instances."""
+    client = _async_executor.get_nacos_client()
+    return await client.list_instances(param)
+
+
+async def call_deregister_instance(param: DeregisterInstanceParam) -> bool:
+    """Deregister instance."""
+    client = _async_executor.get_nacos_client()
+    return await client.deregister_instance(param)
+
+
+async def call_subscribe(param: SubscribeServiceParam) -> None:
+    """Subscribe to service."""
+    client = _async_executor.get_nacos_client()
+    await client.subscribe(param)
+
+
+async def call_unsubscribe(param: SubscribeServiceParam) -> None:
+    """Unsubscribe from service."""
+    client = _async_executor.get_nacos_client()
+    await client.unsubscribe(param)
+
+
+async def call_list_services(param: ListServiceParam):
+    """List services."""
+    client = _async_executor.get_nacos_client()
+    return await client.list_services(param)
+
+
+async def call_register_instance(param: RegisterInstanceParam) -> None:
+    """Register instance."""
+    client = _async_executor.get_nacos_client()
+    await client.register_instance(param)
+
+
+def _cleanup_async_executor():
+    """Cleanup the async executor."""
+    try:
+        _async_executor.shutdown()
+    except Exception as e:
+        plugin_logger.error(f"Error during async executor cleanup: {e}")
+
+
+# Register cleanup function
+atexit.register(_cleanup_async_executor)

framework/fit/python/plugin/fit_py_nacos_registry/service/config.py

@@ -0,0 +1,134 @@
+# -- encoding: utf-8 --
+# Copyright (c) 2025 Huawei Technologies Co., Ltd. All Rights Reserved.
+# This file is a part of the ModelEngine Project.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# ======================================================================================================================
+"""
+Configuration module for Nacos registry server.
+"""
+from v2.nacos import ClientConfigBuilder
+
+from fitframework import value
+from fitframework.utils import tools
+
+
+@value('registry-center.server.addresses', converter=tools.to_list)
+def get_registry_server_addresses() -> list:
+    """Get the list of registry server addresses."""
+    pass
+
+
+@value('nacos.username', default_value=None)
+def get_nacos_username() -> str:
+    """
+    Get the Nacos username.
+    
+    Returns:
+        Nacos username.
+    """
+    pass
+
+
+@value('nacos.password', default_value=None)
+def get_nacos_password() -> str:
+    """
+    Get the Nacos password.
+    
+    Returns:
+        Nacos password.
+    """
+    pass
+
+
+@value('nacos.accessKey', default_value=None)
+def get_nacos_access_key() -> str:
+    """
+    Get the Nacos access key.
+    
+    Returns:
+        Nacos access key.
+    """
+    pass
+
+
+@value('nacos.secretKey', default_value=None)
+def get_nacos_secret_key() -> str:
+    """
+    Get the Nacos secret key.
+    
+    Returns:
+        Nacos secret key.
+    """
+    pass
+
+
+@value('nacos.namespace', default_value="")
+def get_nacos_namespace() -> str:
+    """
+    Get the Nacos namespace.
+    
+    Returns:
+        Nacos namespace.
+    """
+    pass
+
+
+@value('nacos.isEphemeral', default_value=True, converter=bool)
+def get_heartbeat_is_ephemeral() -> bool:
+    """
+    Get whether the heartbeat is ephemeral.
+    
+    Returns:
+        Whether the heartbeat is ephemeral.
+    """
+    pass
+
+
+@value('nacos.heartBeatInterval', default_value=5000, converter=int)
+def get_heartbeat_interval() -> int:
+    """
+    Get the heartbeat interval in milliseconds.
+    
+    Returns:
+        Heartbeat interval in milliseconds.
+    """
+    pass
+
+
+@value('nacos.heartBeatTimeout', default_value=15000, converter=int)
+def get_heartbeat_timeout() -> int:
+    """
+    Get the heartbeat timeout in milliseconds.
+    
+    Returns:
+        Heartbeat timeout in milliseconds.
+    """
+    pass
+
+
+@value('nacos.weight', default_value=1.0, converter=float)
+def get_heartbeat_weight() -> float:
+    """
+    Get the heartbeat weight.
+    
+    Returns:
+        Heartbeat weight.
+    """
+    pass
+
+
+def build_nacos_config():
+    """
+    Build the Nacos client configuration.
+    
+    Returns:
+        Configured Nacos client config.
+    """
+    return (ClientConfigBuilder()
+            .server_address(get_registry_server_addresses()[0])
+            .namespace_id(get_nacos_namespace() or 'local')
+            .username(get_nacos_username())
+            .password(get_nacos_password())
+            .access_key(get_nacos_access_key())
+            .secret_key(get_nacos_secret_key())
+            .build())