优秀开源项目kombu源码分析之registry和entrypoint

YAML 开源 源码分析   2017-03-22 14:27:38 发布
您的评价:
     
0.0
收藏     0收藏
文件夹
标签
(多个标签用逗号分隔)

我曾经在一些公众场合说过心中的优秀Python开发者。Flask和Requests的作者就不说了,21世纪最缺的就是idea,他们不仅有而且还都用非常优美的方式做出来了。另外我还提到了Celery作者Ask Solem,并不是因为Celery很有名它的主要作者就优秀了,我对ask的欣赏,完全是看Celery及其相关依赖的源代码的时候产生的。

有多年后台开发的工程师想必清楚,Celery本身涉及到的技术点其实在业界应用是很广泛的。Celery能这么流行,我们先排除没有进行技术深入下的盲从,和它诞生的非常早以外,我认为这和项目的内部设计的非常好也是有关的。

接下来的几篇文章我将分析Celery使用的Kombu库中的一些设计实现让大家对这个优秀项目更了解,并从中学习可扩展开发的实践。

Kombu是什么?

当一个项目变得越来越复杂,就要考虑只保留核心,并把其他部分分拆到不同的项目中以便减少未来的维护和开发的成本。Flask、IPython都是这样做的。

Kombu是一个把消息传递封装成统一接口的库。 Celery一开始先支持的RabbitMQ,也就是使用AMQ协议。由于要支持越来越多的消息代理,但是这些消息代理是不支持AMQ协议的,需要一个东西把所有的消息代理的处理方式统一起来,甚至可以理解为把它们「伪装成支持AMQ协议」。Kombu的最初的实现叫做carrot, 后来经过重构才成了Kombu。

registry

registry也就是「注册」,有按需加入的意思,在Python标准库和一些优秀开源项目中都有应用。我们先看个django的 场景 ,为了减少篇幅我没有列出CheckRegistry类中其他方法:

### source code start
from itertools import chain


classCheckRegistry:

    def__init__(self):
        self.registered_checks = []
        self.deployment_checks = []

    defregister(self, check=None, *tags, **kwargs):
        kwargs.setdefault('deploy', False)

        definner(check):
            check.tags = tags
            if kwargs['deploy']:
                if check not in self.deployment_checks:
                    self.deployment_checks.append(check)
            elif check not in self.registered_checks:
                self.registered_checks.append(check)
            return check

        if callable(check):
            return inner(check)
        else:
            if check:
                tags += (check, )
            return inner

    deftag_exists(self, tag, include_deployment_checks=False):
        return tag in self.tags_available(include_deployment_checks)

    deftags_available(self, deployment_checks=False):
        return set(chain(*[check.tags for check in self.get_checks(deployment_checks) if hasattr(check, 'tags')]))

    defget_checks(self, include_deployment_checks=False):
        checks = list(self.registered_checks)
        if include_deployment_checks:
            checks.extend(self.deployment_checks)
        return checks


registry = CheckRegistry()
register = registry.register
tag_exists = registry.tag_exists

### source code end
@register('mytag', 'another_tag')
defmy_check(apps, **kwargs):
    pass


print tag_exists('another_tag')
print tag_exists('not_exists_tag')

可以看到每次用registry.register都能动态的添加新的tag,最后还用 register = registry.register 这样的方式列了个别名。执行结果如下:

❯ python django_example.py
True
False

kombu库包含对消息的序列化和反序列化工作的实现,可以同时支持多种序列化方案,如pickle、json、yaml和msgpack。假如你从前没有写过这样可扩展的项目,可能想的是每种的方案的loads和dumps都封装一遍,然后用一个大的if/elif/else来控制最后的序列化如何执行。

那么在kombu里面是怎么用的呢?我简化下它的 实现 :

import codecs
from collections import namedtuple

codec = namedtuple('codec', ('content_type', 'content_encoding', 'encoder'))

classSerializerNotInstalled(Exception):
    pass


classSerializerRegistry(object):
    def__init__(self):
        self._encoders = {}
        self._decoders = {}
        self._default_encode = None
        self._default_content_type = None
        self._default_content_encoding = None

    defregister(self, name, encoder, decoder, content_type,
                 content_encoding='utf-8'):
        if encoder:
            self._encoders[name] = codec(
                content_type, content_encoding, encoder,
            )
        if decoder:
            self._decoders[content_type] = decoder

    def_set_default_serializer(self, name):
        try:
            (self._default_content_type, self._default_content_encoding,
             self._default_encode) = self._encoders[name]
        except KeyError:
            raise SerializerNotInstalled(
                'No encoder installed for {0}'.format(name))

    defdumps(self, data, serializer=None):
        if serializer and not self._encoders.get(serializer):
            raise SerializerNotInstalled(
                'No encoder installed for {0}'.format(serializer))

        if not serializer and isinstance(data, unicode):
            payload = data.encode('utf-8')
            return 'text/plain', 'utf-8', payload

        if serializer:
            content_type, content_encoding, encoder = \
                self._encoders[serializer]
        else:
            encoder = self._default_encode
            content_type = self._default_content_type
            content_encoding = self._default_content_encoding

        payload = encoder(data)
        return content_type, content_encoding, payload

    defloads(self, data, content_type, content_encoding):
        content_type = (content_type if content_type
                        else 'application/data')
        content_encoding = (content_encoding or 'utf-8').lower()

        if data:
            decode = self._decoders.get(content_type)
            if decode:
                return decode(data)
        return data


registry = SerializerRegistry()
dumps = registry.dumps
loads = registry.loads
register = registry.register

其实kombu还实现了unregister限于篇幅我就不展开了。现在我们想添加yaml的支持,只需要加这样一个函数:

defregister_yaml():
    try:
        import yaml
        registry.register('yaml', yaml.safe_dump, yaml.safe_load,
                          content_type='application/x-yaml',
                          content_encoding='utf-8')
    except ImportError:

        defnot_available(*args, **kwargs):
            """Raise SerializerNotInstalled.
            Used in case a client receives a yaml message, but yaml
            isn't installed.
            """
            raise SerializerNotInstalled(
                'No decoder installed for YAML. Install the PyYAML library')
        registry.register('yaml', None, not_available, 'application/x-yaml')


register_yaml()

这样就支持yaml了。如果希望默认使用yaml来序列化,可以执行:

registry._set_default_serializer('yaml')

是不是非常好扩展,如果哪天我希望去掉对pickle(安全问题),就可以直接注释对应的函数就好了。写个小例子试验下:

yaml_data = """\
float: 3.1415926500000002
int: 10
list: [george, jerry, elaine, cosmo]
string: The quick brown fox jumps over the lazy dog
unicode: "Th\\xE9 quick brown fox jumps over th\\xE9 lazy dog"
"""

content_type, content_encoding, payload = dumps(yaml_data, serializer='yaml')
print content_type, content_encoding

assert loads(payload, content_type=content_type, content_encoding=content_encoding) == yaml_data

运行的结果就是:

❯ python kombu_example.py
application/x-yaml utf-8

entrypoint

在我的书里面介绍过如果使用标准库自带的pkg_resources.iter_entry_points实现一个简单的插件系统。这在kombu上面也有应用,在序列化实现模块的最后加了这么几句:

from pkg_resources import iter_entry_points

for ep in iter_entry_points('kombu.serializers'):
    args = ep.load()
    register(ep.name, *args)

这是什么东西呢?pkg_resources是一个用于包发现和资源访问的模块,我们可以实现不同的kombu扩展,如果在这个扩展项目的setup.py里面设置对应的entry_points,在安装之后,运行上述代码的时候就会自动找到这些扩展,并注册进来。这就是一个扩展系统。Flake8就是最好的这个扩展玩法的范例。

kombu的扩展不多,我选择 kombu-fernet-serializers 来进行介绍。首先看一下它的setup.py文件:

...
    entry_points={
        'kombu.serializers': [
            'fernet_json = kombu_fernet.serializers.json:register_args',
            'fernet_yaml = kombu_fernet.serializers.yaml:register_args',
            'fernet_pickle = kombu_fernet.serializers.pickle:register_args',
            'fernet_msgpack = kombu_fernet.serializers.msgpack:register_args',
        ]
    }
...

注意到了吧,这个entry点就是kombu.serializers,安装之后就多了4个序列化方案,我们看一下fernet_json的实现:

import anyjson as _json

from . import fernet_encode, fernet_decode

MIMETYPE = 'application/x-fernet-json'

register_args = (
    fernet_encode(_json.dumps),
    fernet_decode(_json.loads),
    MIMETYPE,
    'utf-8',
)

而fernet_yaml也被放进了模块的方式,其实和在函数内殊途同归:

rom kombu.exceptions import SerializerNotInstalled

from . import fernet_encode, fernet_decode

try:
    import yaml
except ImportError:
    defnot_available(*args, **kwargs):
        """In case a client receives a yaml message, but yaml
        isn't installed."""
        raise SerializerNotInstalled(
            'No decoder installed for YAML. Install the PyYAML library')

    yaml_encoder = not_available
    yaml_decoder = None
else:
    yaml_encoder = yaml.safe_dump
    yaml_decoder = yaml.safe_load

MIMETYPE = 'application/x-fernet-yaml'

register_args = (
    fernet_encode(yaml_encoder),
    fernet_decode(yaml_decoder) if yaml_decoder else None,
    MIMETYPE,
    'utf-8',
)

事实上,我们并不需要了解fernet_encode和fernet_decode是如何对消息做对称加密的,只是感受下这样添加扩展的方式是不是很优雅呢?

 

来自:http://www.dongwm.com/archives/优秀开源项目kombu源码分析之registry/

 

扩展阅读

玩转docker镜像和镜像构建
部署私有Docker Registry
Docker在PHP项目开发环境中的应用
从Docker Hub和docker-registry看优秀的后端服务设计实现
开源容器集群管理系统Kubernetes架构及组件介绍

为您推荐

PHP生成二维码的类及方法
C/C++ 框架,类库,资源集合
机器学习框架、库和软件集合:awesome-machine-learning
机器学习开源项目、类库、软件集合
awesome-machine-learning - 非常棒的机器学习框架,库和软件集合

更多

YAML
开源
源码分析
Python开发
相关文档  — 更多
相关经验  — 更多
相关讨论  — 更多