Python依赖注入容器的实现原理与设计模式

学习搭子，你好！在前文中咱们学习了依赖注入的基本概念和应用，但你可能好奇：那些强大的依赖注入容器到底是怎么工作的？它们如何管理复杂的依赖关系？今天咱们就深入底层，一起揭开依赖注入容器的神秘面纱！

1. 1. 服务注册（Service Registration）
• • 告诉容器：当需要某个类型时，应该创建什么对象
• • 支持多种注册方式：类型、实例、工厂函数、单例等
2. 2. 依赖解析（Dependency Resolution）
• • 自动分析类型的构造函数参数
• • 递归解析所有依赖关系
• • 处理循环依赖等复杂情况
3. 3. 生命周期管理（Lifecycle Management）
• • 控制对象的创建和销毁时机
• • 支持不同的生命周期：瞬时、单例、作用域等
4. 4. 配置与扩展（Configuration & Extension）
• • 支持配置文件或代码配置
• • 提供扩展点允许自定义行为

# 挑战1：动态类型系统
# Python没有编译时类型检查，依赖关系在运行时才确定
defsome_function(service):  # service是什么类型？只有运行时才知道
return service.do_something()

# 挑战2：鸭子类型
# 依赖关系基于协议而非类型
classDuck:
defquack(self):
return"Quack!"

defmake_it_quack(duck_like_object):  # 不需要继承特定接口
return duck_like_object.quack()

# 挑战3：装饰器与元编程
# Python支持强大的元编程能力，容器需要与之协同
@inject
defprocess_data(service: IService = Provide[CONTAINER.service]):
# 装饰器修改了函数签名
pass

# simple_container.py
import inspect
from typing importDict, Type, Any, Callable, Union

classSimpleContainer:
"""简易依赖注入容器"""

def__init__(self):
# 注册表：类型 -> 实现
self._registry: Dict[Type, Union[Type, Any, Callable]] = {}
# 单例缓存
self._singletons: Dict[Type, Any] = {}

defregister(self, interface: Type, implementation=None):
"""注册服务"""
if implementation isNone:
# 如果没有提供实现，假定接口本身就是实现类
self._registry[interface] = interface
else:
self._registry[interface] = implementation

defresolve(self, interface: Type) -> Any:
"""解析服务"""
if interface notinself._registry:
raise ValueError(f"未注册的服务: {interface}")

        implementation = self._registry[interface]

# 如果已经是实例，直接返回
ifnot inspect.isclass(implementation) andnotcallable(implementation):
return implementation

# 创建实例
returnself._instantiate(implementation)

def_instantiate(self, cls_or_factory) -> Any:
"""实例化类或调用工厂函数"""
ifcallable(cls_or_factory) andnot inspect.isclass(cls_or_factory):
# 是工厂函数
return cls_or_factory(self)

# 是类，需要解析构造函数参数
        signature = inspect.signature(cls_or_factory.__init__)
        parameters = {}

for param_name, param in signature.parameters.items():
if param_name == 'self':
continue

# 获取参数的类型注解
            param_type = param.annotation
if param_type == inspect.Parameter.empty:
raise ValueError(f"构造函数参数'{param_name}'缺少类型注解")

# 递归解析依赖
            parameters[param_name] = self.resolve(param_type)

return cls_or_factory(**parameters)

# 定义接口和实现
classIDatabase:
defquery(self, sql: str) -> list:
pass

classMySQLDatabase(IDatabase):
def__init__(self, host: str, port: int):
self.host = host
self.port = port

defquery(self, sql: str) -> list:
return [{"id": 1, "name": "示例数据"}]

classUserService:
def__init__(self, db: IDatabase):
self.db = db

defget_user(self, user_id: int):
returnself.db.query(f"SELECT * FROM users WHERE id = {user_id}")

# 使用容器
container = SimpleContainer()
container.register(IDatabase, MySQLDatabase)
container.register(UserService)

# 注意：MySQLDatabase需要配置参数，我们的容器还不支持！
# 这里会报错，因为MySQLDatabase需要host和port参数

# container_with_params.py
from typing importDict, Any, Union, Type, Callable, Optional
import inspect

classContainerWithParams(SimpleContainer):
"""支持构造函数参数注入的容器"""

def__init__(self):
super().__init__()
# 参数配置：类型 -> 参数字典
self._parameters: Dict[Type, Dict[str, Any]] = {}
# 配置值：类型+参数名 -> 值
self._config_values: Dict[str, Any] = {}

defregister_with_params(self, interface: Type, implementation=None, **params):
"""注册服务并指定构造函数参数"""
super().register(interface, implementation)
if params:
self._parameters[interface] = params

defset_config(self, interface: Type, param_name: str, value: Any):
"""设置配置值"""
        key = f"{interface.__name__}.{param_name}"
self._config_values[key] = value

def_instantiate(self, cls_or_factory) -> Any:
"""改进的实例化方法，支持参数注入"""
ifcallable(cls_or_factory) andnot inspect.isclass(cls_or_factory):
return cls_or_factory(self)

# 获取类
        cls = cls_or_factory

# 检查是否有预定义的参数
if cls inself._parameters:
# 直接使用预定义参数
return cls(**self._parameters[cls])

# 解析构造函数参数
        signature = inspect.signature(cls.__init__)
        parameters = {}

for param_name, param in signature.parameters.items():
if param_name == 'self':
continue

# 检查是否有配置值
            config_key = f"{cls.__name__}.{param_name}"
if config_key inself._config_values:
                parameters[param_name] = self._config_values[config_key]
continue

# 获取类型注解
            param_type = param.annotation
if param_type == inspect.Parameter.empty:
# 没有类型注解，检查是否有默认值
if param.default != inspect.Parameter.empty:
                    parameters[param_name] = param.default
else:
raise ValueError(f"参数'{param_name}'缺少类型注解且无默认值")
else:
# 递归解析依赖
                parameters[param_name] = self.resolve(param_type)

return cls(**parameters)

container = ContainerWithParams()

# 注册MySQLDatabase并指定参数
container.register_with_params(
    IDatabase,
    MySQLDatabase,
    host='localhost',
    port=3306
)

# 注册UserService
container.register(UserService)

# 现在可以正确解析了
user_service = container.resolve(UserService)
print(user_service.get_user(1))  # 正常工作

1. 1. 不支持单例模式
2. 2. 不支持工厂函数
3. 3. 不能处理复杂的依赖图

# advanced_container.py
from enum import Enum
from typing importDict, Any, Union, Type, Callable, Optional, List
import inspect

classLifecycle(Enum):
"""对象生命周期"""
    TRANSIENT = "transient"# 每次解析都创建新实例
    SINGLETON = "singleton"# 全局单例
    SCOPED = "scoped"# 作用域单例

classRegistration:
"""服务注册信息"""

def__init__(
        self,
        interface: Type,
        implementation: Union[Type, Any, Callable],
        lifecycle: Lifecycle = Lifecycle.TRANSIENT,
        factory: Optional[Callable] = None,
        instance: Optional[Any] = None
):
self.interface = interface
self.implementation = implementation
self.lifecycle = lifecycle
self.factory = factory
self.instance = instance  # 对于单例，存储实例

classAdvancedContainer:
"""高级依赖注入容器"""

def__init__(self):
self._registrations: Dict[Type, Registration] = {}
self._scoped_instances: Dict[Type, Any] = {}
self._resolution_stack: List[Type] = []  # 用于检测循环依赖
self._parent: Optional['AdvancedContainer'] = None

# ---------- 注册方法 ----------

defregister(
        self,
        interface: Type,
        implementation=None,
        lifecycle: Lifecycle = Lifecycle.TRANSIENT
) -> 'AdvancedContainer':
"""基本注册"""
if implementation isNone:
            implementation = interface

self._registrations[interface] = Registration(
            interface=interface,
            implementation=implementation,
            lifecycle=lifecycle
        )
returnself

defregister_singleton(
        self,
        interface: Type,
        implementation=None
) -> 'AdvancedContainer':
"""注册单例"""
returnself.register(interface, implementation, Lifecycle.SINGLETON)

defregister_instance(
        self,
        interface: Type,
        instance: Any
) -> 'AdvancedContainer':
"""注册已有实例"""
self._registrations[interface] = Registration(
            interface=interface,
            implementation=type(instance),
            lifecycle=Lifecycle.SINGLETON,
            instance=instance
        )
returnself

defregister_factory(
        self,
        interface: Type,
        factory: Callable
) -> 'AdvancedContainer':
"""注册工厂函数"""
self._registrations[interface] = Registration(
            interface=interface,
            implementation=None,
            lifecycle=Lifecycle.TRANSIENT,
            factory=factory
        )
returnself

# ---------- 解析方法 ----------

defresolve(self, interface: Type) -> Any:
"""解析服务"""
# 检查循环依赖
if interface inself._resolution_stack:
raise RuntimeError(f"检测到循环依赖: {' -> '.join([t.__name__ for t in self._resolution_stack] + [interface.__name__])}")

self._resolution_stack.append(interface)

try:
# 查找注册信息
            registration = self._find_registration(interface)

# 根据生命周期处理
if registration.lifecycle == Lifecycle.SINGLETON:
returnself._resolve_singleton(registration)
elif registration.lifecycle == Lifecycle.SCOPED:
returnself._resolve_scoped(registration)
else:  # TRANSIENT
returnself._resolve_transient(registration)
finally:
self._resolution_stack.pop()

def_find_registration(self, interface: Type) -> Registration:
"""查找注册信息"""
# 先在当前容器查找
if interface inself._registrations:
returnself._registrations[interface]

# 然后在父容器查找
ifself._parent:
returnself._parent._find_registration(interface)

raise ValueError(f"未注册的服务: {interface}")

def_resolve_singleton(self, registration: Registration) -> Any:
"""解析单例"""
if registration.instance isnotNone:
return registration.instance

# 创建实例并缓存
        instance = self._create_instance(registration)
        registration.instance = instance
return instance

def_resolve_scoped(self, registration: Registration) -> Any:
"""解析作用域单例"""
if registration.interface inself._scoped_instances:
returnself._scoped_instances[registration.interface]

        instance = self._create_instance(registration)
self._scoped_instances[registration.interface] = instance
return instance

def_resolve_transient(self, registration: Registration) -> Any:
"""解析瞬时对象"""
returnself._create_instance(registration)

def_create_instance(self, registration: Registration) -> Any:
"""创建实例"""
if registration.factory:
# 使用工厂函数
return registration.factory(self)

# 检查是否已经是实例
ifnot inspect.isclass(registration.implementation):
return registration.implementation

# 解析构造函数
returnself._instantiate_class(registration.implementation)

def_instantiate_class(self, cls: Type) -> Any:
"""实例化类"""
# 获取所有构造函数参数
        signature = inspect.signature(cls.__init__)
        kwargs = {}

for param_name, param in signature.parameters.items():
if param_name == 'self':
continue

# 检查参数类型
            param_type = param.annotation

if param_type != inspect.Parameter.empty:
# 有类型注解，尝试解析
try:
                    kwargs[param_name] = self.resolve(param_type)
except (ValueError, RuntimeError):
# 无法解析，检查是否有默认值
if param.default != inspect.Parameter.empty:
                        kwargs[param_name] = param.default
else:
raise
else:
# 没有类型注解，使用默认值
if param.default != inspect.Parameter.empty:
                    kwargs[param_name] = param.default
else:
# 既没有类型注解也没有默认值，尝试作为值参数
# 在实际应用中，这里可能需要更复杂的逻辑
pass

return cls(**kwargs)

# ---------- 容器组合 ----------

defcreate_child_container(self) -> 'AdvancedContainer':
"""创建子容器"""
        child = AdvancedContainer()
        child._parent = self
return child

defenter_scope(self) -> 'AdvancedContainer':
"""进入新的作用域"""
        scoped_container = self.create_child_container()
# 作用域容器继承父容器的注册
return scoped_container

defexit_scope(self):
"""退出作用域（清理作用域实例）"""
self._scoped_instances.clear()

# 使用示例
container = AdvancedContainer()

# 1. 注册单例数据库连接
container.register_singleton(IDatabase, MySQLDatabase)

# 2. 注册需要配置的类
classConfigService:
def__init__(self, app_name: str = "默认应用"):
self.app_name = app_name

container.register(ConfigService)

# 3. 注册工厂函数
defcreate_logger(container: AdvancedContainer) -> 'Logger':
    config = container.resolve(ConfigService)
return Logger(name=config.app_name)

container.register_factory(Logger, create_logger)

# 4. 使用作用域
with container.enter_scope() as scoped_container:
# 在作用域内解析的服务会缓存
    db1 = scoped_container.resolve(IDatabase)
    db2 = scoped_container.resolve(IDatabase)
print(db1 is db2)  # True（在作用域内是同一个实例）

# 作用域结束，作用域实例被清理

classContainerComposite:
"""容器组合模式实现"""

def__init__(self):
self._children: List['ContainerComposite'] = []

defadd_child(self, child: 'ContainerComposite'):
self._children.append(child)

defresolve(self, interface: Type) -> Any:
# 首先尝试自己解析
if interface inself._registrations:
returnsuper().resolve(interface)

# 然后尝试子容器
for child inself._children:
try:
return child.resolve(interface)
except ValueError:
continue

raise ValueError(f"未找到服务: {interface}")

definject(func):
"""依赖注入装饰器"""

    @wraps(func)
defwrapper(*args, **kwargs):
# 获取函数签名
        signature = inspect.signature(func)
        bound_args = signature.bind_partial(*args, **kwargs)

# 填充缺失的参数
for param_name, param in signature.parameters.items():
if param_name in bound_args.arguments:
continue

# 检查类型注解
            param_type = param.annotation
if param_type != inspect.Parameter.empty:
# 从容器解析
                container = get_current_container()
                bound_args.arguments[param_name] = container.resolve(param_type)
elif param.default != inspect.Parameter.empty:
                bound_args.arguments[param_name] = param.default

return func(*bound_args.args, **bound_args.kwargs)

return wrapper

classLifecycleStrategy:
"""生命周期策略接口"""

defget_instance(
        self,
        registration: Registration,
        container: 'AdvancedContainer'
) -> Any:
pass

classSingletonStrategy(LifecycleStrategy):
"""单例策略"""

def__init__(self):
self._instance = None

defget_instance(self, registration, container):
ifself._instance isNone:
self._instance = container._create_instance(registration)
returnself._instance

classTransientStrategy(LifecycleStrategy):
"""瞬时策略"""

defget_instance(self, registration, container):
return container._create_instance(registration)

# 在容器中使用策略
classStrategyBasedContainer:
def__init__(self):
self._strategies: Dict[Lifecycle, LifecycleStrategy] = {
            Lifecycle.SINGLETON: SingletonStrategy(),
            Lifecycle.TRANSIENT: TransientStrategy()
        }

# 直接循环依赖
classA:
def__init__(self, b: 'B'):
self.b = b

classB:
def__init__(self, a: A):
self.a = a

# 间接循环依赖
classX:
def__init__(self, y: 'Y'):
self.y = y

classY:
def__init__(self, z: 'Z'):
self.z = z

classZ:
def__init__(self, x: X):
self.x = x

defdetect_circular_dependency(container, start_type: Type) -> bool:
"""使用深度优先搜索检测循环依赖"""

    visited = set()
    recursion_stack = set()

defdfs(current_type: Type) -> bool:
if current_type in recursion_stack:
returnTrue# 发现循环

if current_type in visited:
returnFalse

        visited.add(current_type)
        recursion_stack.add(current_type)

# 获取当前类型的所有依赖
        dependencies = get_dependencies(container, current_type)

for dep_type in dependencies:
if dfs(dep_type):
returnTrue

        recursion_stack.remove(current_type)
returnFalse

return dfs(start_type)

defget_dependencies(container, interface: Type) -> List[Type]:
"""获取类型的所有依赖"""
    registration = container._find_registration(interface)
    cls = registration.implementation

ifnot inspect.isclass(cls):
return []

    signature = inspect.signature(cls.__init__)
    dependencies = []

for param in signature.parameters.values():
if param.name == 'self':
continue

        param_type = param.annotation
if param_type != inspect.Parameter.empty:
            dependencies.append(param_type)

return dependencies

classA:
def__init__(self):
self.b = None

defset_b(self, b: 'B'):
self.b = b

classB:
def__init__(self, a: A):
self.a = a

# 使用
a = A()
b = B(a)
a.set_b(b)  # 后设置属性

classLazyProxy:
"""延迟代理"""

def__init__(self, container, interface: Type):
self._container = container
self._interface = interface
self._instance = None

def__getattr__(self, name):
ifself._instance isNone:
self._instance = self._container.resolve(self._interface)
returngetattr(self._instance, name)

classA:
def__init__(self, b: LazyProxy):
self.b = b  # b是代理，使用时才真正解析

classB:
def__init__(self, a: A):
self.a = a

# 将循环依赖拆分为多个接口
classIReader:
defread(self) -> str:
pass

classIWriter:
defwrite(self, data: str):
pass

classA(IReader, IWriter):
def__init__(self, reader: IReader, writer: IWriter):
self.reader = reader
self.writer = writer

defread(self):
returnself.reader.read()

defwrite(self, data):
self.writer.write(data)

classReflectionCache:
"""反射结果缓存"""

    _signature_cache: Dict[Type, inspect.Signature] = {}
    _dependency_cache: Dict[Type, List[Type]] = {}

    @classmethod
defget_signature(cls, obj) -> inspect.Signature:
        key = obj if inspect.isclass(obj) elsetype(obj)

if key notin cls._signature_cache:
            cls._signature_cache[key] = inspect.signature(
                obj if inspect.isclass(obj) else obj.__init__
            )

return cls._signature_cache[key]

    @classmethod
defget_dependencies(cls, clazz: Type) -> List[Type]:
if clazz notin cls._dependency_cache:
            signature = cls.get_signature(clazz)
            dependencies = []

for param in signature.parameters.values():
if param.name == 'self':
continue

                param_type = param.annotation
if param_type != inspect.Parameter.empty:
                    dependencies.append(param_type)

            cls._dependency_cache[clazz] = dependencies

return cls._dependency_cache[clazz]

classPrecompiledContainer:
"""预编译依赖图的容器"""

def__init__(self):
self._dependency_graph: Dict[Type, List[Type]] = {}
self._resolution_plan: Dict[Type, List[Type]] = {}

defcompile(self):
"""编译所有注册的服务"""
for interface inself._registrations:
self._compile_type(interface)

def_compile_type(self, interface: Type):
"""编译单个类型"""
if interface inself._resolution_plan:
return

        registration = self._find_registration(interface)

# 获取直接依赖
        dependencies = ReflectionCache.get_dependencies(registration.implementation)
self._dependency_graph[interface] = dependencies

# 递归编译依赖
for dep in dependencies:
self._compile_type(dep)

# 生成解析顺序（拓扑排序）
self._resolution_plan[interface] = self._topological_sort(interface)

def_topological_sort(self, start_type: Type) -> List[Type]:
"""拓扑排序生成解析顺序"""
        visited = set()
        result = []

defdfs(node: Type):
if node in visited:
return

            visited.add(node)

for neighbor inself._dependency_graph.get(node, []):
                dfs(neighbor)

            result.append(node)

        dfs(start_type)
return result

classLightweightService:
"""使用__slots__的轻量级服务"""

    __slots__ = ('data', 'cache', 'config')

def__init__(self, data, cache, config):
self.data = data
self.cache = cache
self.config = config

# 相比普通类，每个实例节省约50%内存

# dependency_injector的核心类（简化版）
classProvider:
"""所有提供者的基类"""

def__init__(self):
self._overrides = []

def__call__(self):
"""获取提供的对象"""
raise NotImplementedError

defoverride(self, provider):
"""重写提供者"""
self._overrides.append(provider)
returnself

classFactory(Provider):
"""工厂提供者"""

def__init__(self, provides, *args, **kwargs):
super().__init__()
self._provides = provides
self._args = args
self._kwargs = kwargs

def__call__(self):
ifself._overrides:
returnself._overrides[-1]()

# 解析依赖
        resolved_args = [resolve_arg(arg) for arg inself._args]
        resolved_kwargs = {k: resolve_arg(v) for k, v inself._kwargs.items()}

returnself._provides(*resolved_args, **resolved_kwargs)

classSingleton(Factory):
"""单例提供者"""

def__init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._instance = None

def__call__(self):
ifself._instance isNone:
self._instance = super().__call__()
returnself._instance

classContainerMeta(type):
"""容器的元类，用于自动装配"""

def__new__(mcs, name, bases, namespace):
# 收集所有Provider实例
        providers = {}
for key, value in namespace.items():
ifisinstance(value, Provider):
                providers[key] = value

# 创建容器类
        cls = super().__new__(mcs, name, bases, namespace)
        cls._providers = providers

return cls

classContainer(metaclass=ContainerMeta):
"""依赖注入容器"""

    @classmethod
defprovider(cls, name):
"""获取提供者"""
return cls._providers[name]

defwire(self, modules=None):
"""自动装配"""
# 遍历模块，注入依赖
pass

classProvide:
"""依赖描述符"""

def__init__(self, provider):
self._provider = provider

def__get__(self, instance, owner):
if instance isNone:
returnself

# 从容器获取实例
        container = getattr(instance, '_container', None)
if container isNone:
raise RuntimeError("容器未设置")

return container.provider(self._provider)()

definject(func):
"""依赖注入装饰器"""

    @wraps(func)
defwrapper(self, *args, **kwargs):
# 解析参数依赖
        signature = inspect.signature(func)
        bound = signature.bind_partial(self, *args, **kwargs)

for param_name, param in signature.parameters.items():
if param_name in bound.arguments:
continue

# 检查是否有Provide描述符
ifisinstance(param.default, Provide):
                bound.arguments[param_name] = param.default.__get__(self, type(self))

return func(*bound.args, **bound.kwargs)

return wrapper

# test_container.py
import unittest
from typing import Protocol

classIService(Protocol):
defdo_something(self) -> str:
pass

classServiceImpl(IService):
defdo_something(self) -> str:
return"完成"

classTestContainer(unittest.TestCase):

defsetUp(self):
self.container = AdvancedContainer()

deftest_register_and_resolve(self):
"""测试基本注册和解析"""
self.container.register(IService, ServiceImpl)

        instance = self.container.resolve(IService)

self.assertIsInstance(instance, ServiceImpl)
self.assertEqual(instance.do_something(), "完成")

deftest_singleton_lifecycle(self):
"""测试单例生命周期"""
self.container.register_singleton(IService, ServiceImpl)

        instance1 = self.container.resolve(IService)
        instance2 = self.container.resolve(IService)

self.assertIs(instance1, instance2, "单例应该返回同一个实例")

deftest_transient_lifecycle(self):
"""测试瞬时生命周期"""
self.container.register(IService, ServiceImpl)

        instance1 = self.container.resolve(IService)
        instance2 = self.container.resolve(IService)

self.assertIsNot(instance1, instance2, "瞬时应该返回不同实例")

deftest_circular_dependency_detection(self):
"""测试循环依赖检测"""
classA:
def__init__(self, b: 'B'):
self.b = b

classB:
def__init__(self, a: A):
self.a = a

self.container.register(A)
self.container.register(B)

withself.assertRaises(RuntimeError) as context:
self.container.resolve(A)

self.assertIn("循环依赖", str(context.exception))

deftest_factory_registration(self):
"""测试工厂函数注册"""
defcreate_service(container) -> IService:
return ServiceImpl()

self.container.register_factory(IService, create_service)

        instance = self.container.resolve(IService)
self.assertIsInstance(instance, ServiceImpl)

# test_integration.py
classTestIntegration(unittest.TestCase):

deftest_complete_application(self):
"""测试完整应用"""
        container = AdvancedContainer()

# 配置容器
        container.register_singleton(IDatabase, MySQLDatabase)
        container.register(IUserRepository, UserRepository)
        container.register(IOrderRepository, OrderRepository)
        container.register(UserService)
        container.register(OrderService)
        container.register(OrderController)

# 模拟Web请求
        controller = container.resolve(OrderController)
        result = controller.create_order({
"user_id": 123,
"items": [{"product_id": 1, "quantity": 2}]
        })

self.assertTrue(result["success"])
self.assertIn("order_id", result["data"])

deftest_container_hierarchy(self):
"""测试容器层级"""
        root = AdvancedContainer()
        child = root.create_child_container()

# 在根容器注册
        root.register(IService, ServiceImpl)

# 子容器应该能解析父容器的服务
        instance = child.resolve(IService)
self.assertIsInstance(instance, ServiceImpl)

# 子容器注册的服务不应该影响父容器
classAnotherService(IService):
defdo_something(self):
return"另一个"

        child.register(IService, AnotherService)

# 父容器还是原来的实现
        root_instance = root.resolve(IService)
self.assertIsInstance(root_instance, ServiceImpl)

# 子容器是自己的实现
        child_instance = child.resolve(IService)
self.assertIsInstance(child_instance, AnotherService)

# test_performance.py
import time
import statistics

classTestPerformance(unittest.TestCase):

deftest_resolution_speed(self):
"""测试解析速度"""
        container = AdvancedContainer()

# 注册100个服务
for i inrange(100):
classService:
def__init__(self, num: int = i):
self.num = num

            container.register(Service)

# 测量解析时间
        times = []
for _ inrange(1000):
            start = time.perf_counter()
            container.resolve(Service)
            end = time.perf_counter()
            times.append((end - start) * 1000)  # 转换为毫秒

        avg_time = statistics.mean(times)
        max_time = max(times)

print(f"平均解析时间: {avg_time:.3f}ms")
print(f"最大解析时间: {max_time:.3f}ms")

# 断言性能要求
self.assertLess(avg_time, 1.0, "平均解析时间应小于1ms")
self.assertLess(max_time, 5.0, "最大解析时间应小于5ms")

1. 1. 容器的核心职责：注册、解析、生命周期管理
2. 2. 实现关键技术：反射、递归解析、循环依赖检测
3. 3. 设计模式应用：组合、装饰器、策略模式
4. 4. 性能优化技巧：缓存、预编译、内存优化
5. 5. 测试策略：单元测试、集成测试、性能测试

• • 理解胜于记忆：知道容器如何工作，比记住API更重要
• • 逐步构建：复杂系统从简单开始，逐步添加功能
• • 模式识别：识别并应用设计模式解决通用问题
• • 性能意识：在实现功能的同时考虑性能影响

1. 1. 研究其他语言实现：学习Java Spring、C# Autofac等容器的设计
2. 2. 深入元编程：研究Python描述符、元类在DI中的应用
3. 3. 探索AOP集成：了解如何将面向切面编程与依赖注入结合
4. 4. 研究容器标准：了解JSR-330等依赖注入标准

1. 1. 自己实现一次：按照今天的步骤，自己动手实现一个容器
2. 2. 阅读开源代码：深入研究dependency-injector等框架的源码
3. 3. 应用到实际项目：在项目中实践分层架构和依赖注入
4. 4. 参与开源贡献：为现有的依赖注入框架贡献代码或文档