memcached分布式缓存服务器实现，含代码演示

1.简介

开源的高性能分布式缓存系统，以BSD License授权协议发布，用于加速动态WEB应用程序，减轻数据库负载。通过在内存中缓存数据来减少数据库读取的次数，提高动态网站的速度。

它简单而强大，部署方便，使用快捷，支持客户端语言丰富。

图示

1.1软件特点

数据存储方式为简单的key/value,数据格式包含图像、视频、数据库等。

实现逻辑一半在客户端一半在服务端.

Server服务器各自独立.

基于C/S架构，协议简单，配置简单.

客户端API支持广泛，包含C/C++/C#/Java/PHP/Perl/Python/Ruby等.

CPU要求不高，但是内存要求较高，速度快稳定好.

每个数据可以配置超时时间，超时都会过期，用户查找该数据会返回NULL.

缺乏认证和安全管制.

1.2 存储机制

1.2.1 memcached术语

chunk：为固定大小的内存空间。

Page：对应实际的物理空间，1个page为1M，分配给slab之后，切分成chunk。

Slab：同样大小的chunk也称为slab。

Slab class：特定大小的chunk组，由相同大小的chunk组成。

1.2.2 存储机制

Memcached将内存分成众多chunk，这是存储用户数据的最小单位。相同大小的chunk组成slab，slab是存储管理的单位，每个slab对应一个列表，记录哪些chunk已经被分配。Memcached根据用户数据大小，将其放入比它大的最小slab中。

图例

因为chunk的大小是固定的长度内存，所以memcached的存储也存在资源浪费。如将100字节的数据缓存到128字节的chunk中，剩余的字节就浪费了。

图例

1.3 基本安装

1.3.1 安装

yum install memcached -y

1.3.2 配置文件

cat /etc/sysconfig/memcached

1.3.3 服务控制

systemctl <start|restart|stop> memcached

2 Shell维护

2.1 调试方法

telnet localhost 11211

2.2 存储命令

<command name> <key> <flags> <exptime><bytes>[noreply]<data block>

key:不超过250个字符的字符串，不能包含空格和控制字符串，支持二进制。

flags:数据标签，可以自定义。

exptime:失效时间，可以是相对时间（秒数）或者绝对时间（UNIX时间表示），设置为0表示永不生效。

bytes:值数据的字节数，字节数可以设置为0

noreply:可选，表示不需要server回复报文。

2.2.1 set

set 改写或新增

2.2.2 add

add 只新增

2.2.3 replace

replace 只改写

2.2.4 append

append 从后面增添数据

2.2.5 prepend

prepend 在前面插入数据

2.2.6 cas

cas 检查并改写，在改写之前，保证在自己上次读取后数据项没有发生变化

2.3 读取/删除命令

get <key>*

delete <key>[noreply]

2.3.1 get

get 由key得到返回数据，包括flag,value长度和value

gets 根据cas值获取数据的get，意味着必须有特定的cas值，否则不会有结果

2.3.2 delete

delete 删除

2.4 数据加减

2.4.1 incr

incr <key><value>[noreply]

<key>：键

<value>:增加或者减少的数值，为64位无符号整数

2.4.2 decr

decr <key><value>[noreply]

<key>：键

<value>:增加或者减少的数值，为64位无符号整数

2.5 管理命令

2.5.1 stats

stats [setting|sizes|…]

stats items 以slab为单位列出数据项的信息,显示各个slab中item的数目和存储时长（最后一次访问距离现在的秒数）

stats slabs 以slab为单位列出数据项的信息，更关注于slab的性能。

stats sizes 用于确定多大的slab能有更好的存储效率，返回2列，第一列是item的大小，第二列是item的个数。

stats settings 查看设置情况

2.5.2 touch

touch <key><exptime>[noreply]

用于改变value的超时时间，不能在value已经过期使用。

2.5.3 flush_all

flush_all[delay][noreply]

使所有数据项都超时

<exptime>:失效时间

<delay>:延时秒数

2.5.4 version

version

显示版本信息

2.5.5 quit

quit

退出

3 API模拟

3.1 接口（python）

3.1.1 安装库

pip install python-memcached –y

3.1.2 接口汇总

@set(key,val,time=0,min_compress_len=0)

无条件键值对的设置，其中的time用于设置超时，单位是秒，而min_compress_len则用于设置zlib压缩(注:zlib是提供数据压缩用的函式库)

@set_multi(mapping,time=0,key_prefix='',min_compress_len=0)

设置多个键值对，key_prefix是key的前缀，完整的键名是key_prefix+key, 使用方法如下

>>> mc.set_multi({'k1' : 1, 'k2' : 2}, key_prefix='pfx_') == []

>>> mc.get_multi(['k1', 'k2', 'nonexist'], key_prefix='pfx_') == {'k1' : 1, 'k2' : 2}

@add(key,val,time=0,min_compress_len=0)

添加一个键值对，内部调用_set()方法

@replace(key,val,time=0,min_compress_len=0)

替换value，内部调用_set()方法

@get(key)

根据key去获取value，出错返回None

@get_multi(keys,key_prefix='')

获取多个key的值，返回的是字典。keys为key的列表

@delete(key,time=0)

删除某个key。time的单位为秒，用于确保在特定时间内的set和update操作会失败。如果返回非0则代表成功

@incr(key,delta=1)

自增变量加上delta，默认加1，使用如下

>>> mc.set("counter", "20")

>>> mc.incr("counter")

@decr(key,delta=1)

自减变量减去delta，默认减1

3.1.3 python编码

3.1.3.1 编码

#!/usr/bin/env python

import memcache

mc = memcache.Client(['127.0.0.1:11211'],debug=0)

mc.set("foo","bar")

value = mc.get("foo")

print value

3.1.3.2 测试

图示

3.2 加速mysql

3.2.1 库安装

yum install mariadb-server mariadb-devel gcc gcc-devel python-devel -y

pip install mysql-python

3.2.2 MySQL配置

CREATE DATABASEmemcached;

GRANT ALLPRIVILEGES ON memcached.* TO 'admin'@'node01' IDENTIFIED BY 'admin';

GRANT ALLPRIVILEGES ON memcached.* TO 'admin'@'192.168.8.211' IDENTIFIED BY 'admin';

GRANT ALLPRIVILEGES ON memcached.* TO 'admin'@'%' IDENTIFIED BY 'admin';

create tablelun(

id int(8) not null primary keyauto_increment,

pool_name char(20) not null,

lun_name char(20) not null,

group_id int(8) not null default 1

);

insert intolun(pool_name,lun_name,group_id)values("pool01","lun01",80);

insert intolun(pool_name,lun_name,group_id)values("pool01","lun02",80);

insert intolun(pool_name,lun_name,group_id)values("pool01","lun03",80);

nsert intolun(pool_name,lun_name,group_id)values("pool02","lun01",80);

insert intolun(pool_name,lun_name,group_id)values("pool02","lun02",80);

insert intolun(pool_name,lun_name,group_id)values("pool02","lun03",80);

insert intolun(pool_name,lun_name,group_id)values("pool03","lun01",90);

insert intolun(pool_name,lun_name,group_id)values("pool03","lun02",90);

insert intolun(pool_name,lun_name,group_id)values("pool03","lun03",90);

3.2.3 python编码

3.2.3.1 编码

import MySQLdb

import memcache

memc = memcache.Client(['node01:11211'], debug=1);

try:

conn = MySQLdb.connect (host = "node01",

user = "admin",

passwd = "admin",

db = "memcached")

except MySQLdb.Error, e:

print "Error %d: %s" % (e.args[0], e.args[1]) sys.exit (1)

popularfilms = memc.get('lun')

#print popularfilms

if not popularfilms:

cursor = conn.cursor()

cursor.execute('select pool_name,lun_name from lun order by id desc limit 10')

rows = cursor.fetchall()

memc.set('lun',rows,60)

print "Updated memcached with MySQL data"

else:

print "Loaded data from memcached"

for row in popularfilms:

print "%s, %s" % (row[0], row[1])

3.2.3.2 测试

图示

4 memcached 分布式

前面说到memcached的特性中就有“实现逻辑一半在客户端一半在服务端”，客户端的工作是怎样选择哪台服务器进行读或者写，如果服务器无法连接应该怎么切换；服务端的工作是怎样存储数据，怎样分配内存。

所以memcached的分布式，可以说是基于客户端的分布式。

4.1 架构分布式

负载均衡器以基于源方式将业务流量负载至5台中间件服务器，这些中间件服务器也是memcached服务器，每个memcached服务器之间相互独立，memcached缓存了该中间件上的业务数据。如果业务运行中，有一台中间件（memcached服务器）故障，这时通过负载均衡器可以分发新流量至其他中间件服务器，其它4台memcached只需要缓存故障的一台memcached服务器的缓存数据。

这种方式不需要额外的客户端代码实现，也不需要第三方软件支持，并且memcache服务器故障影响面小；但是它也会造成环境复杂化变高，对中间件服务器的要求增高等缺点。

4.2 架构模块化

图示

架构模块化中，将memcached服务器做成统一集群，提供给中间件服务器存储业务全局缓存。客户端使用分布式算法将缓存的数据存放于不同的memcached服务器中。

4.2.1 余数计算算法

4.2.1.1 实现逻辑

就是根据服务器台数的余数进行分散。求得键的整数哈希值，再除以服务器台数，根据其余数来选择服务器。就是hash(key)%sessions.size()。这种方法简单，数据分散也相当优秀，默认的python-mecached客户端算法就是余数计算算法。

但是其缺点也是很致命的，就是在添加和删除服务器时，缓存重组的代价相当巨大。添加/删除服务器后，余数就会产生巨变，这样就无法保证获取时计算的服务器节点与保存时相同，从而影响缓存命中率，造成原有的缓存数据将大规模失效，在短时间内极大的压力可能会压爆数据库。

为了直观一点，现举一个例子，如场景中有3(srv_NUM)台服务器，插入12个记录，hash(key)分别为：0,1,2,3,4,5,6,7,8,9,10,11 .服务器分布计算方法为srvID = hash(key)%srv_NUM

所以根据计算得到分布如下：

Srv0: 0,3,6,9

Srv1:1,4,7,10

Srv2:2,5,8,11

当增加一个srv3之后，按照同样的算法得出分布如下：

Srv0:0,4,8

Srv1:1,5,9

Srv2:2,6,10

Srv3:3,7,11

如果这样就可以看出，数据分布和增加节点之前相差很大。

4.2.1.2 代码实现

代码实现较为简单，在memcache.Client中增加服务器即可。

import MySQLdb

import memcache

memc =memcache.Client(['node01:11211','node02:11211'], debug=1);

try:

conn = MySQLdb.connect (host ="node01",

user ="admin",

passwd ="admin",

db ="memcached")

except MySQLdb.Error, e:

print "Error %d: %s" %(e.args[0], e.args[1])

sys.exit (1)

popularfilms = memc.get('lun01')

#print popularfilms

if not popularfilms:

cursor = conn.cursor()

cursor.execute('select pool_name,lun_namefrom lun order by id desc limit 10')

rows = cursor.fetchall()

memc.set('lun01',rows,600)

memc.set('lun02',rows,600)

memc.set('lun03',rows,600)

memc.set('lun04',rows,600)

print "Updated memcached with MySQLdata"

else:

print "Loaded data frommemcached"

for row in popularfilms:

print "%s, %s" % (row[0], row[1])

4.2.1.3 测试验证

4.2.1.3.1 一般验证

缓存分布

使用4.2.1.2中代码进行测试，数据超时时间设置为10分钟，存入4组数据，查看数据在各个memcached中的分布情况。

运行脚本，确认数据已经缓存。

图示

登录node01服务器查看缓存情况

图示

登录node02服务器查看缓存情况

图示

由实验可以看出在node01中缓存了lun01,lun03,lun04的数据，node02中缓存了lun02的数据。

节点失效影响大

运行脚本，确认数据已经缓存

图示

选取一个缓存少量数据的节点，停止memcached服务

根据观察node02只缓存了lun02一条数据，所以将node02的memcached服务停止

图示

从python客户端中读取缓存数据

客户端循环读出所有缓存的数据，代码如下：

import sys

import MySQLdb

import memcache

memc = memcache.Client(['node01:11211','node02:11211'], debug=1);

try:

conn = MySQLdb.connect (host = "node01",

user = "admin",

passwd = "admin",

db = "memcached")

except MySQLdb.Error, e:

print "Error %d: %s" % (e.args[0], e.args[1])

sys.exit (1)

for i in '1234':

popularfilms = memc.get('lun0'+i)

print "Loaded data from memcached"

for row in popularfilms:

print "%s, %s" % (row[0], row[1])

运行读脚本，发现只能读出一条缓存记录，其他3条缓存记录失效。

图示

此时重新缓存数据，再次运行读脚本，发现4条记录都可以读出。

图示

4.2.1.3.2 大并发验证

大并发验证环节，我们使用4台memcached服务器作为集群。首先使用3台memcached服务器，客户端模拟1W条key/value的插入数据，统计每个memcached服务器的缓存分布次数。通过增加/移除验证整体缓存命中率。

插入1W条key/value数据记录

#!/usr/bin/env python

import memcache

import pylibmc

def getClient(ports):

#ports = '12'

srvs = []

for i in ports:

#srvs.append(('127.0.0.1:%d'%(i+20000), 100))

srvs.append('node0'+i+':11211')

mc = None

#print srvs

#print '+++++++'

if use_pylibmc:

mc = pylibmc.Client(srvs,

binary=True,

behaviors={"tcp_nodelay": True, "ketama": True})

else:

mc = memcache.Client(srvs, debug =1)

return mc

def createData(ports,data):

mc = getClient(ports)

for j in range(0,data):

mc.set(str(j),str(j))

mc.disconnect_all()

print "the Data have been created!"

def hitsize(ports,data):

mc = getClient(ports)

hit = 0

for l in range(0,data):

if mc.get(str(l)) != None:

hit = hit + 1

# print "node0%s hit is %f"%(k,hit)

return hit

mc.disconnect_all()

if __name__ == '__main__':

use_pylibmc = False

createData('123',10000)

print "node01 hitsize is:%f"%(hitsize('1',10000))

print "node02 hitsize is:%f"%(hitsize('2',10000))

print "node03 hitsize is:%f"%(hitsize('3',10000))

图示

添加服务器

图示

删除服务器

图示

从测试中发现，余数计算算法的缓存分布还是很均匀的，但是在增加/删除服务器时命中率下降特别严重，命中率只有30%左右。

4.2.2 一致性hash

4.2.2.1 实现逻辑

首先求出memcached 服务器（节点）的哈希值，并将其配置到0～232 的圆（continuum）上。然后用同样的方法求出存储数据的键的哈希值，并映射到圆上。然后从数据映射到的位置开始顺时针查找，将数据保存到找到的第一个服务器上。如果超过232 仍然找不到服务器，就会保存到第一台memcached 服务器上。

图示

从上图的状态中添加一台memcached 服务器。余数分布式算法由于保存键的服务器会发生巨大变化，而影响缓存的命中率，但一致性哈希中，只有在continuum 上增加服务器的地点逆时针方向的第一台服务器上的键会受到影响。

图示

因此，一致性哈希最大限度地抑制了键的重新分布。而且，有的一致性哈希的实现方法还采用了虚拟节点的思想。使用一般的hash函数的话，服务器的映射地点的分布非常不均匀。因此，使用虚拟节点的思想，为每个物理节点（服务器）在continuum上分配100～200 个点。这样就能抑制分布不均匀，最大限度地减小服务器增减时的缓存重新分布。

其实一致性哈希的应用特别的广泛，就拿分布式存储举例，华为fusionStorage分布式存储的DHT算法，其实就是一致性哈希的算法实现，ceph的crush算法等等。如果说的白话一点，就是一致性哈希算法是做了2次哈希，一次对对象，一次对主机。将对象的hash值映射到0-2的32次方的圆环中，在将主机的hash值映射到0-2的32次方的圆环中。按照顺时针决定对象存储的主机。所以这样处理不管是添加还是删除节点，数据的变动只是一个扇区，不会发生很大的变化。为了解决平衡性（节点数较小的情况下），可以引入虚拟节点，一个主机对应若干多个的虚拟节点，使原来在圆环中的一个主机点变为多个主机点，就类似于原来是hash(node01)变成了hash(node01-1);hash(node01-2)。这样数据就可以更加均衡了。

4.2.2.2 代码实现

4.2.2.2.1 一致性哈希

本例中使用pylibmc来进行代码实现。pylibmc依赖于libmemcached,libmemcached是用C语言编写的memcached客户端，它支持异步和同步传输、一致性哈希分布、大对象的支持、本地复制等。本例中使用了ketama哈希一致性算法。有详细了解的同学请移步（http://libmemcached.org/libMemcached.html）本例中使用tar包编译安装，过程省略。

pip installpylibmc

插入数据并统计各个节点缓存次数

#!/usr/bin/env python

import memcache

import pylibmc

def getClient(ports):

#ports = '12'

srvs = []

for i in ports:

#srvs.append(('127.0.0.1:%d'%(i+20000),100))

srvs.append('node0'+i+':11211')

mc = None

#print srvs

#print '+++++++'

if use_pylibmc:

mc = pylibmc.Client(srvs,

binary=True,

behaviors={"tcp_nodelay": True, "ketama": True})

else:

mc = memcache.Client(srvs, debug =1)

return mc

def createData(ports,data):

mc = getClient(ports)

for j in range(0,data):

mc.set(str(j),str(j))

mc.disconnect_all()

print "the Data have beencreated!"

def hitsize(ports,data):

mc = getClient(ports)

hit = 0

for l in range(0,data):

if mc.get(str(l)) != None:

hit = hit + 1

# print "node0%s hit is %f"%(k,hit)

return hit

mc.disconnect_all()

if __name__ == '__main__':

use_pylibmc = True

createData('123',10000)

print "node01 hitsize is:%f"%(hitsize('1',10000))

print "node02 hitsizeis:%f"%(hitsize('2',10000))

print"node03 hitsize is:%f"%(hitsize('3',10000))

增加Server统计全局命中次数

if __name__ == '__main__':

use_pylibmc = True

# createData('123',10000)

print "total hitsize when okis:%f"%(hitsize('123',10000))

print"total hitsize when add server is:%f"%(hitsize('1234',10000))

减少Server统计全局命中次数

if __name__ == '__main__':

use_pylibmc = True

# createData('123',10000)

print "total hitsize when okis:%f"%(hitsize('123',10000))

print"total hitsize when del server is:%f"%(hitsize('12',10000))

4.2.2.2.2 增加虚拟节点

在实现逻辑中也有简单描写虚拟节点的介绍，在这里在介绍一下。余数计算法的分布还是很均匀的，但是cache命中率就太差了，我们就用到了一致性哈希来解决这个问题，但是经过我们的测试发现，一致性哈希虽然解决了命中率问题，但是一致性哈希缓存分布相比余数计算法还是比较差强人意的，特别是在服务器节点少的情况下，具体数据请参考测试验证篇章。而且我们也想到了因为集群是分布式的，所以服务器的配置可能是不一样的，如果这样，是不是用一致性哈希会导致更加大的分布不均匀现象呢？

所以引入了虚拟节点，其实也比较简单，在一致性哈希的介绍中我们了解到会对主机地址做一次哈希运算，虚拟节点的意思就是将原来的一个主机地址转变成了多个，这样一个主机就可以取到多个hash值，然后在圆环中占据多个位置。对于配置不同的主机我们也有了解决方案，配置高的，我们增加更多的虚拟节点；而对于配置低的，控制少量的虚拟节点。

4.2.2.3 测试验证

4.2.2.3.1 一致性哈希

本例的编码和余数计算法中大并发测试类似，统计每个memcached服务器的命中次数，并和余数计算法进行对比。然后模拟一台memcached服务器添加/删除，统计出整体失效的命中率。

插入1W条key/value数据记录

图示

添加服务器

图示

删除服务器

图示

从测试中发现，一致性哈希算法的缓存分布相对于余数计算法稍许逊色了一些，但是在增加/删除服务器时命中率提高了很多，综合命中率可以达到70%左右。

4.3 第三方软件实现

4.3.1 简介

本章中列举的第三方软件为magent,magent是一个简单，但是非常有用的memcached缓存服务器的代理小程序。它虽然小，但是功能比较合理，支持常用的memcached指令，基于libevent事件驱动库，备份集群，和每个memcache服务器保持多个长连接，关键它还支持ketama算法。

图示

4.3.2 安装

4.3.2.1 编译安装

tar –xzvfmagent-0.5.tar.gz

sed -i"s#LIBS = -levent#LIBS = -levent -lm#g" Makefile

vim ketama.h(在开头加入)

#ifndef SSIZE_MAX

#define SSIZE_MAX 32767

#endif

make

cp magent /usr/bin/

4.3.2.2 启动运行

magent –k -uroot -l 192.168.8.211 -p 11210 -s 192.168.8.211:11211 -s 192.168.8.212:11211 -s192.168.8.213:11211

4.3.3 测试

4.3.3.1 命中率测试

录入数据，查看各个节点缓存分布

#!/usr/bin/env python

import memcache

import pylibmc

def getClient(ports):

#ports = '12'

srvs = []

for i in ports:

#srvs.append(('127.0.0.1:%d'%(i+20000), 100))

srvs.append('node0'+i+':11210')

mc = None

#print srvs

#print '+++++++'

if use_pylibmc:

mc = pylibmc.Client(srvs,

binary=True,

behaviors={"tcp_nodelay": True, "ketama": True})

else:

mc = memcache.Client(srvs, debug =1)

return mc

def createData(ports,data):

mc = getClient(ports)

for j in range(0,data):

mc.set(str(j),str(j))

mc.disconnect_all()

print "the Data have been created!"

def hitsize(ports,data):

mc = getClient(ports)

hit = 0

for l in range(0,data):

if mc.get(str(l)) != None:

hit = hit + 1

# print "node0%s hit is %f"%(k,hit)

return hit

mc.disconnect_all()

def hitnodesize(ports,p,data):

mc = memcache.Client(['node0'+ports+':'+p],debug=1)

hit = 0

for l in range(0,data):

if mc.get(str(l)) != None:

hit = hit + 1

return hit

mc.disconnect_all()

if __name__ == '__main__':

use_pylibmc = False

createData('1',10000)

print "node01 hitsize is:%f"%(hitnodesize('1','11211',10000))

print "node02 hitsize is:%f"%(hitnodesize('2','11211',10000))

print "node03 hitsize is:%f"%(hitnodesize('3','11211',10000))

图示

增加节点

图示

减少节点

图示

经过测试发现因为magent使用了ketama算法，和上例中的一致性哈希算法一致，所以命中率也基本差不多，在70%左右。

4.3.3.2 高可用测试

本次测试和命中率测试方法一样，主要关注减少节点的命中率，如果命中率100%则高可用成功；如果是其它数值，则高可用测试失败。

录入数据，查看各节点缓存分布

代码沿用命中率测试代码，初始使用2台memcached服务器，1台buckup服务器。