缓存系统 Memcached

绝大多数的 Web 应用都把数据存在数据库里面，应用服务器从中读取数据。随着数据量和访问量的增加，频繁通过直接访问数据库获得数据的方式对数据库造成很大的负担。Memcached 是一个高性能的分布式内存对象缓存系统，用在动态 Web 应用中减轻数据库负载。它通过在内存中缓存数据来减少读取数据库的次数，从而提高 Web 应用的速度。

Memcached 的守护进程使用 C 编写，客户端则可以使用任何语言，通过 Memcached 协议与守护进程通信。常见的 Python 客户端有如下几种。

1．Python-memcached：纯 Python 实现的客户端，但是只实现了 Memcached 的基本操作功能。

2．Pymemcache：Pinterest 开源的纯 Python 实现的客户端，比 Python-memcached 要快。

3．Python-Libmemcached：豆瓣开源的使用 Cython 实现的 Libmemcached 的客户端。Libmemcached 是 Memcached 的 C/C++客户端，它支持一致性哈希、分布式、同步/异步传输等功能。豆瓣用的 Libmemcached 是打过补丁的。

4．Pylibmc：它也是 Libmemcached 的一个使用 C 编写的 Python 封装的客户端。

5．Libmc（http://bit.ly/28SHgfG ）：它是使用 C++和 Cython 编写的一个轻量高效的 Memcached Python 客户端，支持以一致性哈希环的方式一次性与多个 Memcached 节点交互。Libmc 使用了类似 Pylibmc（http://bit.ly/28W6bn7 ）的基准测试方法（http://bit.ly/28S8wuh ），对比了 Pylibmc、Python-memcached 和 Libmc 的性能，测试结果可以在豆瓣的 Travis-CI 集成（http://bit.ly/28SkeVh ）中找到。可以看到在所有的测试中都是 Libmc 最快。豆瓣在生产环境中已经完全使用 Libmc 替换了 Libmemcached 加 Python-Libmemcached 的组合。原因是 Libmemcached 代码非常复杂且 bug 多，豆瓣所使用的分支也与上游脱节，不可维护。

Libmc 安装配置

首先安装 Memcached：

＞ sudo apt-get install memcached

安装之后 Memcached 已经自动启动了：

＞ ps-ef|grep memcached
memcache 3108    1 0 16:22 ?      00:00:00/usr/bin/memcached-m 64-p 11211-u
    memcache-l 127.0.0.1

这是默认的启动方式，表示使用 memcache 这个用户，最大占用 64 MB 内存，监听本机的回路地址和 11211 端口。默认只使用 64 MB 内存实在太小了，在生产环境中需要根据实际内存情况使用更大的值。下面会使用 Memcached 分布式处理缓存，所以再启动两个监听其他端口的实例：

＞ /usr/bin/memcached-m 64-p 11212-u memcache-l 127.0.0.1-d
＞ /usr/bin/memcached-m 64-p 11213-u memcache-l 127.0.0.1-d

接着安装 Libmc。如果使用 Vagrant 的方式，编译需要的内存将超出虚拟机所能提供的大小，需要创建 swap 分区用户替代内存资源：

＞ sudo dd if=/dev/zero of=/swapfile bs=64M count=48
＞ sudo mkswap/swapfile
＞ sudo swapon/swapfile

这样，我们就可以借用 3 GB 硬盘作为内存来使用了。Docker 方式不需要使用 swap 分区。

＞ pip install libmc

我们使用 Libmc 的配置如下（mc.py）：

from libmc import (
    Client, MC_HASH_MD5, MC_POLL_TIMEOUT, MC_CONNECT_TIMEOUT, MC_RETRY_TIMEOUT
)

mc=Client(
    [
        'localhost',
        'localhost:11212',
        'localhost:11213 mc_213'
    ],
    do_split=True,
    comp_threshold=0,
    noreply=False,
    prefix=None,
    hash_fn=MC_HASH_MD5,
    failover=False
)

mc.config(MC_POLL_TIMEOUT, 100)
mc.config(MC_CONNECT_TIMEOUT, 300)
mc.config(MC_RETRY_TIMEOUT, 5)

我们解析下这个获得 mc 实例的程序。

• libmc.Client 接受的第一个参数就是 Memcached 服务器的列表，格式是“host-name[:port] [alias]”。其中端口和别名是可选项，若未指定端口，则默认端口为 11211，如 localhost:11211 和 localhost 是等价的。
• do_split：默认值为 False，会拒绝大于 1 MB 的值的存储；如果设置为 True，小于 10 MB 的值会被切分成多个块，但是不能存储大于 10 MB 的值。
• comp_threshold：所有类型的值都会被编码为字符串，如果设置的这个阈值等于 0，字符串不会使用 zlib 压缩；如果这个字符串的长度大于设置的阈值且不为 0，就会使用 zlib 压缩。默认值是 0。
• noreply：表示是否开启 noreply 模式。在这种 noreply 模式下，更新缓存的 set 操作可以不需要 Memcached 服务端响应，这使得 set 操作非常快。默认值为 False。
• prefix：缓存键的前缀，常用于区分不同的环境中相同的缓存键。
• hash_fn：对键做哈希的函数标识，可选值有 MC_HASH_MD5、MC_HASH_FNV1_32、MC_HASH_FNV1A_32 和 MC_HASH_CRC_32。默认值是 MC_HASH_MD5。
• failover：表示当前服务器不可用时，是否做故障转移到写一个服务器。默认值为 False。
• MC_POLL_TIMEOUT：执行 Memcached 的 set/get 操作的超时时间，单位为 ms。
• MC_CONNECT_TIMEOUT：连接 Memcached 的超时时间，单位为 ms。
• MC_RETRY_TIMEOUT：当 Memcached 服务器不可用等情况下，重试链接的时间，单位为 s。这种重试一直持续到服务恢复。

使用原生 SQL 缓存

现在参考豆瓣缓存服务客户端 douban-mc（http://bit.ly/28Y4bsO ）实现一个写原生 SQL 的 Flask 应用的例子。

首先实现一个叫作 cache 的装饰器，它可以方便地在方法上定义缓存键和缓存时间（mc_decorator.py）。cache 需要一个格式化的函数，它可以把各种需要缓存的参数格式化成为一个缓存键：

import re


__formaters={}
percent_pattern=re.compile(r'%\w')
brace_pattern=re.compile(r'\{[\w\d\.\[\]_]+\}')

def formater(text):
    percent=percent_pattern.findall(text)
    brace=brace_pattern.search(text)
    if percent and brace:
        raise Exception('mixed format is not allowed')

    if percent:
        n=len(percent)
        return lambda*a,**kw:text%tuple(a[:n])
    elif '%(' in text:
        return lambda*a,**kw:text%kw
    else:
        return text.format


def format(text,*a,**kw):
    f=__formaters.get(text)
    if f is None:
        f=formater(text)
        __formaters[text]=f
    return f(*a,**kw)

format 函数的格式化效果如下：

In:key='web_develop:users:%s'
In:id_=1
In:format(key%'{id_}', id_=id_)
Out:'web_develop:users:1'

In:key='web_develop:users:%s:%s'
In:format(key%('{id_}', '{type}'), id_=id_, type='a')
Out:'web_develop:users:1:a'

inspect 模块提供了一些获取活动对象信息的函数，inspect.getargspec 可以获取函数和方法的参数列表。举两个例子：

In:f=lambda x:x
In:inspect.getargspec(f)
Out:ArgSpec(args=['x'], varargs=None, keywords=None, defaults=None)

In:f=lambda x, y=10:x+y
In:inspect.getargspec(f)
Out:ArgSpec(args=['x', 'y'], varargs=None, keywords=None, defaults=(10,))

通过 getargspec 就可以根据方法的参数获得对应的缓存 key。举个例子：

In:def f(id_, type):
...:   return id_, type
...:

In:arg_names, varargs, varkw, defaults=inspect.getargspec(f)

In:gen_key_factory(key, arg_names, defaults)(1, 2)
Out:('web_develop:users:1:2',{'id_':1, 'type':2})

gen_key_factory 的代码如下：

import inspect


def gen_key_factory(key_pattern, arg_names, defaults):
        args=dict(zip(arg_names[-len(defaults):], defaults)) if defaults else{}
        if callable(key_pattern):
            names=inspect.getargspec(key_pattern)[0]

    def gen_key(*a,**kw):
        aa=args.copy()
        aa.update(zip(arg_names, a))
        aa.update(kw)
        if callable(key_pattern):
            key=key_pattern(*[aa[n] for n in names])
        else:
            key=format(key_pattern,*[aa[n] for n in arg_names],**aa)
        return key and key.replace(' ', '_'), aa
    return gen_key

cache 装饰器通过 gen_key_factory 获得缓存键，在没有缓存的时候 set，缓存未过期前通过 get 使用缓存：

import time
from functools import wraps

MC_DEFAULT_EXPIRE_IN=0 #永不过期


def cache(key_pattern, mc, expire=MC_DEFAULT_EXPIRE_IN, max_retry=0):
    def deco(f):
        arg_names, varargs, varkw, defaults=inspect.getargspec(f)
        if varargs or varkw:
            raise Exception("do not support varargs")
        gen_key=gen_key_factory(key_pattern, arg_names, defaults)
 
        @wraps(f)
        def_(*a,**kw):
            key, args=gen_key(*a,**kw)
            if not key:
                return f(*a,**kw)
            force=kw.pop('force', False)
            r=mc.get(key) if not force else None

            retry=max_retry
            while r is None and retry>0:
                # when node is down, add() will failed
                if mc.add(key+'#mutex', 1, int(max_retry*0.1)):
                    break
                time.sleep(0.1)
                r=mc.get(key)
                retry-=1

            if r is None:
                r=f(*a,**kw)
                if r is not None:
                    mc.set(key, r, expire)
                if max_retry>0:
                    mc.delete(key+'#mutex')
            return r
        _.original_function=f
        return_

    return deco
    def create_decorators(mc):
        def_cache(key_pattern, expire=0, mc=mc, max_retry=0):
            return cache(key_pattern, mc, expire=expire, max_retry=max_retry)
        return{'cache':_cache}

在 mc.py 文件中，添加如下两行：

from decorators import create_decorators
globals().update(create_decorators(mc))

相当于把 cache 放进了 mc 的命名空间中。

基于第 4 章 4.2 节的 Flask-RESTful 例子，将其修改成使用 Libmc 的版本（app_with_mc.py）。User 类的代码如下：

from sqlalchemy import create_engine
from mc import mc, cache

app=Flask(__name__)
DATABASE_URI='mysql://web:web@localhost:3306/r'
api=Api(app)

con=create_engine(DATABASE_URI).connect()

USER_KEY='web_develop:users:%s'


class User(object):
    def__init__(self, id, name, address):
        self.id=id
        self.name=name
        self.address=address

    @classmethod
    def add(cls, name, address):
        sql=('insert into test_user(name, address) '
            'values(%s,%s)')
        id_=con.execute(sql, (name, address)).lastrowid
        cls.clear_mc(id_)
        return cls.get(id_)

    @classmethod
    @cache(USER_KEY%'{id_}') #可以添加第二个参数，设置缓存时间
    def get(cls, id_):
        if not id_:
            return None
        row=con.execute(
        'select id, name, address '
        'from test_user where id=%s', id_).fetchone()
    return cls(*row) if row else None

@classmethod
def get_user_by_name(cls, name):
    sql=('select id from test_user '
    'where name=%s')
    rows=con.execute(sql, name).fetchall()
    return cls.get(*rows[0]) if rows else None

def delete(self):
    con.execute(
        'delete from test_user where id=%s', self.id)
    self.clear_mc(self.id)

@classmethod
def clear_mc(cls, id_):
    mc.delete(USER_KEY%id_)

User 将创建、删除和查询封装起来，还添加了清除缓存、通过 name 字段获取 User 实例这两个方法。需要注意类方法和实例方法的使用：delete 只能在 User 实例后才能调用。

UserResource 相对地也有所修改：

class UserResource(Resource):
    @marshal_with(resource_fields)
    def get(self, name):
        user=User.get_user_by_name(name=name)
        return user

    def put(self, name):
        address=request.form.get('address', '')
        User.add(name, address)
        return{'ok':0}, 201

    def delete(self, name):
        user=User.get_user_by_name(name)
        if user:
            user.delete()
        return{'ok':0}

这样就把 Memcached 引用进来了。

缓存更新策略

有两种常见的更新方案：

• 懒惰式加载。客户端先查询 Memcached，如果命中，则返回结果；如果没命中（没有数据或已经过期），则从数据库中获得最新数据，并写回到 Memcached 中，最后返回结果。这种方法直接、简单。但是在高并发的场景下，突然失效会让后端数据库的压力骤增。
• 主动更新。默认缓存永不失效。当有数据需要更新时，同时也会把最新数据写回到 Memcached 中。这种更新如果耗时过长，应该使用异步的更新，如放在消息队列中。

Memcached 的永不失效其实是设置超时时间为 0，当内存不足时，会触发 LRU 机制，删除最近最少使用的内存空间。

Memcached 使用的经验

1．批量获取时尽可能使用“mc.get_multi”，与同数量的“mc.get”相比，能减少网络请求的次数。

2．对缓存全部更新，可以直接升级缓存键，在缓存键字符串最后加个版本号，比如上例中的 USER_KEY：

USER_KEY='web_develop:users:%s:v2'

3．批量更新缓存的时候应该尽量少给后端数据库带来压力，需要对缓存预热。

4．Memcached 不仅能缓存 SQL 查询的结果，它甚至还可以缓存 HTML 片段。在实际工作中笔者曾经用 Memcached 来缓存 Mako Cache 的页面数据，性能更好。

5．按照 Memcached 协议（http://bit.ly/28UOu5M ），缓存的值必须小于 1 MB，内容可以是任意字符串。虽然 Libmc 通过拆分支持大于 1 MB、小于 10 MB 的结果，但是不鼓励这么做。

6．不鼓励使用读/写/删之外的接口。不鼓励使用的接口包括但不限：append、prepend、incr、decr、add、replace。Memcached 并不适合做上面这些接口做的事，使用这些接口也会给运维带来困难。