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

我曾经在一些公众场合说过心中的优秀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  
  
  
class CheckRegistry:  
  
    def __init__(self):  
        self.registered_checks = []  
        self.deployment_checks = []  
  
    def register(self, check=None, *tags, **kwargs):  
        kwargs.setdefault('deploy', False)  
  
        def inner(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  
  
    def tag_exists(self, tag, include_deployment_checks=False):  
        return tag in self.tags_available(include_deployment_checks)  
  
    def tags_available(self, deployment_checks=False):  
        return set(chain(*[check.tags for check in self.get_checks(deployment_checks) if hasattr(check, 'tags')]))  
  
    def get_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')  
def my_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'))  
  
class SerializerNotInstalled(Exception):  
    pass  
  
  
class SerializerRegistry(object):  
    def __init__(self):  
        self._encoders = {}  
        self._decoders = {}  
        self._default_encode = None  
        self._default_content_type = None  
        self._default_content_encoding = None  
  
    def register(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))  
  
    def dumps(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  
  
    def loads(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的支持，只需要加这样一个函数：

def register_yaml():  
    try:  
        import yaml  
        registry.register('yaml', yaml.safe_dump, yaml.safe_load,  
                          content_type='application/x-yaml',  
                          content_encoding='utf-8')  
    except ImportError:  
  
        def not_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](https://github.com/heroku/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也被放进了模块的方式，其实和在函数内殊途同归：

from kombu.exceptions import SerializerNotInstalled  
  
from . import fernet_encode, fernet_decode  
  
try:  
    import yaml  
except ImportError:  
    def not_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是如何对消息做对称加密的，只是感受下这样添加扩展的方式是不是很优雅呢？

版权声明：本文由 董伟明 原创，未经作者授权禁止任何微信公众号和向掘金(juejin.im)转载，技术博客转载采用 保留署名-非商业性使用-禁止演绎 4.0-国际许可协议
python