张芷铭的个人博客

#Python #注册器 #设计模式 #插件系统

注册器模式通过维护名称到类/函数的映射，实现动态对象创建，适用于插件系统、多算法实现等场景。

核心优势

  • 解耦：模块定义与使用分离
  • 动态：通过字符串名称创建实例
  • 可扩展：新增模块无需修改核心代码

注册器实现

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

class Registry:
    def __init__(self, name):
        self._name = name
        self._obj_dict = {}

    def register(self, name=None, force=False):
        def _register(obj):
            obj_name = name or obj.__name__
            if not force and obj_name in self._obj_dict:
                raise KeyError(f"{obj_name} already registered")
            self._obj_dict[obj_name] = obj
            return obj
        return _register

    def get(self, name):
        if name not in self._obj_dict:
            raise KeyError(f"'{name}' not found. Available: {list(self._obj_dict.keys())}")
        return self._obj_dict[name]

    def list(self):
        return list(self._obj_dict.keys())

# 创建注册器
CONSUMERS = Registry("consumers")
GENERATORS = Registry("generators")

使用示例

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# 注册类
@CONSUMERS.register("flux")
class Flux:
    def __init__(self, config):
        self.config = config

# 注册函数
@GENERATORS.register("single_png")
def single_png_generator(config):
    return {"type": "single_png", "config": config}

# 动态获取
ConsumerClass = CONSUMERS.get("flux")
instance = ConsumerClass({"key": "value"})

配置驱动工厂

1
2
3
4
5
6
7
8

def build_from_config(cfg, registry):
    name = cfg["type"]
    params = cfg.get("args", {})
    obj_class = registry.get(name)
    return obj_class(**params)

config = {"type": "flux", "args": {"bootstrap_servers": "localhost:9092"}}
consumer = build_from_config(config, CONSUMERS)

命令行集成

1
2
3
4
5
6
7

import argparse

parser = argparse.ArgumentParser()
parser.add_argument("--consumer", choices=CONSUMERS.list())
args = parser.parse_args()

ConsumerClass = CONSUMERS.get(args.consumer)

最佳实践

  1. 单一职责：每个注册器管理一种组件类型
  2. 错误提示：列出可用选项
  3. 文档完善：为注册项添加 docstring

Comments