8.缓存与消息队列
8.0. 服务负载较重时，常用的两项基本技术:缓存与消息队列
此前是socket-API及Py中使用基础IP网络，来操作构建通信信道的方式
此后是关于，构建在socket上的特定协议---如何从web获取文档、发送mail、向远程Serv提交命令
8.0.1. 缓存与消息队列有一些共同特点:
8.0.1.1. 使用Memcached/一个消息队列，不是为了实现一个协议来与其他工具进行交互，而是为了编写服务，来解决特定的问题。
8.0.1.2. 这两项技术解决的问题，通常是机构内部特有的问题。通常无法仅从外界就得知一个特定的网站/网络服务使用了哪种缓存、哪种消息队列、哪种负载分配工具
8.0.1.3. HTTP和SMTP这样的工具，都是针对一个特定的负载设定的(HTTP针对超文本文档、SMTP对电子邮件)，但缓存和消息队列，是无需了解所要传输的数据的。
8.1. 使用Memcached:“内存缓存守护进程"，将安装它的Serv的空闲RAM与一个很大的近期最少使用(LRU)的缓存结合使用。可从Memcached学到重要的现代网络概念---分区(sharding)
8.1.1. 使用Memcached实际步骤:
8.1.1.1. 在每台拥有空闲内存的Serv上，都运行一个Memcached守护进程
8.1.1.2. 将所有Memcached守护进程的IP和port列出，将该列表发送给所有要使用Memcached的Cli
8.1.1.3. Cli程序可以访问一个组织级的，速度极快的键值缓存，像是所有Serv间共享的一个巨大的Py字典。该缓存基于LRU。如果有些项长时间没有被访问，就会将这些项丢弃，为新访问的项挪出空间，并记录被频繁访问的项。
http://code.google.com/p/memcached/wiki/Clients列出了许多Memcached的Py-Cli
安装用Py3写的Cli到Py包索引提供的虚拟环境中：
su root
$ pip3 install python3-memcached
>>> import memcache
>>> mc = memcache.Client(['127.0.0.1:11211'])
>>> mc.set('user:19', 'Simple is better than complex.')
0
>>> mc.get('user:19')
'Simple is better than complex.' # 但是我的是没有值
这里的接口与Py-dic类似，将一个str座位值传入set()时，该str会以UTF-8编码直接写入Memcached，再通过get()获取该str时，会进行解码。除了str外，写入任何Py对象，都会自动触发memcache模块的pickle操作，将二进制的pickle存储在Memcached中。只有以str形式存储的值，才可被使用其他语言编写的Cli直接使用
8.1.2. Serv是可以丢弃存储在Memcached中的数据中。Memcached目的是，将重复计算花销较高的结果记录下来，来加速操作。不是用来作为数据的唯一存储区的，如果运行上面的命令时，相应的Memcached正处于繁忙/set()和get()操作间相隔的时间太长，在运行get()的时候，可能回发现str已经超过缓存的有效期，被丢弃了
8-1是Py中使用Memcached的基本模式。进行一个花销很大的整数平方操作前，会先检查Memcached中是否已经保存了之前计算过的答案。如果有，能将答案立刻返回。
# 8-1 squares.py 使用功能Memcached为一个花销很大的操作加速
# !/usr/bin/env python
# -- coding: utf-8 --
import memcache, random, time, timeit
def compute_square(mc, n):
value = mc.get('sq:%d' % n)
if value is None:
time.sleep(0.001) # pretend that computing a square is expensive
value = n * n
mc.set('sq:%d' % n, value)
return value
def main():
mc = memcache.Client(['127.0.0.1:11211'])
def make_request():
compute_square(mc, random.randint(0, 5000))
print('Ten successive runs:')
for i in range(1, 11):
print(' %.2fs' % timeit.timeit(make_request, number=2000), end='')
print()
if __name__ == '__main__':
main()
要运行这个例子，需要再PC的11211-port上运行Memcached。对于最先开始的几百个请求来说，程序会以正常的速度运行。第一次请求计算某个整数的平方时，程序会先发现RAM中没有存储果该整数的平方，必须进行计算。随着程序的运行，发现缓存中已经存储了一些整数的平方，加快了程序的运行速度。
$ python3 squares.py
Ten successive runs:
2.87s 2.94s 1.50s 1.18s 0.95s 0.73s 0.64s 0.56s 0.48s 0.45s
# 但是我的结果是: 7.11s 6.41s 9.51s 14.44s 7.45s 5.91s 10.94s 6.46s 5.74s 6.51s
# 或 4.61s 3.70s 3.64s 3.63s 3.63s 3.76s 3.82s 3.75s 4.76s 3.70s
展示了缓存的普遍特性，随着缓存中的键值对越来越多，程序的运行速度会越来越快。当Memcached存满/所有可能输入都计算过后，速度就不会再有显著提升了
实际程序中，希望将哪种数据写入到缓存中?
8.1.3. 直接将最底层的花销较大的调用存入缓存，如数据库查询、文件系统I/O、对外部服务的查询。
为保证缓存中的数据不过时，需要决定信息在缓存中的保存时间。
8.1.4. 键是必须唯一的。开发者逐渐倾向于使用前缀和编号，来区分存储的各种类型的对象。键最长只能包含250个字符，但通过使用一个强大的散列函数，能对更长的str提供查询功能。Memcached中存储的值可以比键更长，但不能操作1M
8.1.5. Memcached只是一个缓存，存储的内容是暂时的。使用RAM作为介质，一旦重启，会丢失。程序应始终能恢复并重建所有丢失的数据。
8.1.6. 确保缓存返回的数据不能太久，这样返回给用户的数据才是精确的。返回的数据是否太旧，取决于要解决的问题。
3个方法可以用来解决脏数据的问题,确保数据过时后，进行及时清理，永远不返回脏数据:
- Memcached允许为缓存中的每一项设置一个过期时间，到达这个时间，Memcached会负责丢弃这些项
- 如果能建立从信息标识，到缓存中包含该标识的键的映射，就可以再脏数据出现后，主动移除这些缓存项。
- 当缓存中的记录不可用，可重写并使用新内容，替代该条记录，不是简单地移除该记录。这一做法对于每秒能命中好几十次的记录来说很有用。所有使用该缓存的Cli就不会发现，要查找的条目不存在，而会在缓存中重新找到重写过后的记录。同样是这个原因，程序首次启动时，预先安装缓存，对于大型web-site来说，是即为重要的一项技术。
8.1.7. Py的装饰器，可以在不改变函数调用的名称及签名的情况下，对其进行包装，所以它是在Py中增加缓存功能的一种常用做法。Py包索引中，有很多基于Memcached的装饰器缓存库
8.2. 散列与分区
当Memcached-Cli得到包含了多个Memcached实例的列表时，会根据每个键的str值的散列值，对Memcached数据库进行分区(shard)，由计算出的散列值，决定用Memcached集群中的哪台Serv来存储特定的记录。
8-1中，可能会存储的键sq:42和值1764。为了充分利用可用的RAM， Memcached集群会只希望存储一次这个键值对。为加快服务的数据，它希望能在避免冗余的同时，避免不同Serv的协同操作/各Cli间的通信。
意味着，所有Cli除了键和配置的Memcached-Serv列表外，不需要其他任何信息。Cli需要采取某种机制，确定将特定的信息存储在哪台Serv上。
如果没有把相同的键值对映射到同一台Serv上，那么同一个键值对就会被复制多份，减少了可用的总内存空间。Cli在试图移除某Serv上的不可用记录时，其他Serv上仍会存在该记录。
解决方法是，所有Cli都要实现一个相同的稳定算法，将键转换为整数n，根据这个int来从Serv清单选择一台Serv真正进行存储。通过“散列”算法来完成。该算法在构造一个int的时候，会将str的所有位都混合起来，理想状态下，str中的所有模式都会被消除。
8-2中，程序加载的是一个英语单词字典，程序将单词作为键，研究这些单词在4个Serv上的分布规律。
第一个算法视图将字母表分为4个大致平均的部分，并根据单词的首字母来分配键；
另外两个算法使用了散列算法。
8-2 hashing.py 向服务器分配数据的两种机制---数据中的模式与散列值中的位
import hashlib
def alpha_shard(word):
"""Do a poor job of assigning data to servers by using first letters."""
if word[0] < 'g':
return 'server0'
elif word[0] < 'n':
return 'server1'
elif word[0] < 't':
return 'server2'
else:
return 'server3'
def hash_shard(word):
"""Assign data to servers using Python's built-in hash() function."""
return 'server%d' % (hash(word) % d)
def md5_shard(word):
"""Assign data to server using a public hash algorithm."""
data = word.encode('utf-8')
return 'server%d' % (hashlib.md5(data).digest()[-1] % 4)
if __name__ == '__main__':
words = open('/usr/share/dict/words').read().split()
for function in alpha_shard, hash_shard, md5_shard:
d = {'server0': 0, 'server1': 0, 'server2': 0, 'server3':0}
for word in words:
d[function(word.lower())] += 1
print(function.__name__[:-6])
for key, value in sorted(d.items()):
print(' {} {} {:.2}'.format(key, value, value / len(words)))
hash()函数是Py自己内置的散列函数，该函数就是Py字典查找时使用的内部实现，运行速度快，MD5算法复杂，实际上是作为一个用于加密的散列算法已经太弱了，但用于分配Serv的负载还是不错的。
8.2.1. 运行结果中发现，如果用于分配负载的方法直接暴露了数据中的模式，将是相当危险的。
$ python hashing.py
alpha
server0 35285 0.36
...
md5
server0 14777 0.25
1)使用首字母分配负载时，4个Serv负责的首字母数基本相同，但尽管Serv0只负责6个首字母，而Serv3负责7个首字母，Serv0的负载却是Serv3的3倍
2)两个散列算法的表现都非常完美。散列算法没有借助任何与单词有关的模式，都将单词平均分配到了4台Serv上
8.2.2. 如果需要自行设计能自动将负载/数据，分配到集群中节点的服务时，要令多个Cli对同一数据输入给出相同的分配结果，在代码中使用散列算法是十分有用的。
8.3. 消息队列
消息队列协议允许发送可靠的数据块，称之为：消息,而不是数据报(datagram)。数据报是用来特指不可靠服务的，传输过程中，数据可能丢失、重复、重新排列。
8.3.1. 消息队列保证消息的可靠自动传输: 一条消息要么被完好无损地传输至目的地，要么完全不传输。消息队列协议会负责封帧。使用消息队列的Cli从来不需要，在接收到完整的消息前，一直在循环中年不断调用recv()
8.3.2. 消息队列与TCP基于IP传输机制提供的点对点连接不同，使用消息队列的Cli间，可以设置各种各样的拓扑结构。
消息队列可能的应用场景:
8.3.2.1. 当使用email在一个web-site注册新账号时，立刻返回的页面“感谢注册"。用户无需等待，网站通过email服务商传输emal，可能需要好几分钟的时间。web-site的通常做法是，将email地址是放在一个消息队列中，当后台Serv准备好建立一个用于发送的SMTP连接时，直接从消息队列中获取email地址。如果失败，email地址会直接放回队列中，经历更长时间的间隔后重试
8.3.2.2. 可作为自定义远程过程调用(RPC)服务的基础。允许繁忙的前端Serv将一些困难的工作，交给后端Serv来负责。前端Serv可将请求置于消息队列中，几十/几百个后端Serv会对该消息队列进行监听，后端Serv在处理完消息队列中的请求后，会将响应返回给正在等待的前端Serv
8.3.2.3. 经常需要将一些大容量的事件数据，作为小型的有效消息流，集中存储在消息队列汇总并进行分析。一些web-site中，mq已经彻底替代了存储到本地硬盘中的log-os即syslog这样更古老的日志传输机制
8.3.3. MQ程序设计有一个重要特点，具有混合安排并匹配所有Cli与Serv/发布者与订阅者进程的能力。都需连接到同一个MQ-os
8.3.4. MQ的使用，给程序带来了革命性的进步。典型的传统程序，在单个程序中，包含了所有功能。由一层一层的API组成。一个控制thd可能会负责所有API的调用。
如，先从socket读取HTTP数据，然后进行认证，请求解析，调用API进行特定的图像处理，最后将结果写入磁盘中。该控制线thd使用的所有API必须存在于同一台机器上，且被载入到同一个Py运行实例内。
一旦能使用MQ，产生疑惑：
为什么图像处理这样的计算密集型，专业且对网络不可见的工作，要与前端HTTP服务共享CPU和磁盘？
1)可不使用安装了大量不同库的强大机器来构建服务，而使用一些专门用于单一目的的机器，将这些机器集合到集群中，共同提供某个服务。
2)只要负责运维的同事，理解MQ传递的拓扑结构，且保证在进行Serv分离时，没有任何消息丢失，在卸载、安装、重装图像处理Serv时，就完全不会影响位于MQ前端的HTTP服务负载均衡池
所有MQ都支持多种拓扑结构
8.3.4.1. 管道(pipeline)拓扑结构，可能是对于队列的最直观的一种模式:生产者创建消息，将消息提交至队列中，消费者从队列中接受消息。
如，一个照片分享网站的前端Serv，会将用户上传的图片，存储在一个专门用于接收文件的内部队列中。包含许多缩略图生成工具的机房，会从MQ中读取图片。每个图像处理Serv每次从MQ中接收一条消息(消息中包含需要生成缩略图的图片)，为其生成缩略图。站点繁忙时，MQ在运行过程中，可能会越来越长；站点较为空闲时，MQ会变短/再次清空。无论站点是否繁忙，前端Serv都可以直接向等待的Cli返回一个响应，告诉用户，上传已经成功，且很快就能在照片流中看到照片
8.3.4.2. 发布者-订阅者(publisher-subscriber)/扇出(fanout)拓扑结构，看上去和管道差不多，二者的重要区别。pipe的MQ能保证队列中的每个消息，都只会发送给一个消费者(同一张图片发给两台图像Serv很浪费),但sub通常想接受MQ中的所有消息。另一种方法，由sub设置一个过滤器，通过某种特定格式，限定有兴趣消息范围。该类型的MQ可用于，需要向外部世界推送事件的外部服务。
Serv机房可使用这种队列系统，来对哪些系统启动了、因为维护而关闭进行通知。此外，还可在其他MQ创建和销毁的时候，发布它们的IP
8.3.4.3. 请求-响应(request-reply)模式，最为复杂的模式。原因在于消息需要进行往返。前两种模式中，消息生产者的工作非常少。
生产者连接到MQ，然后发送消息。但是，发起请求的MQ-Cli需要保持连接，并等待接收响应。为了支持这一点，MQ必须提供某种寻址机制，从上千个已经连接，且仍然处于等待的Cli中，找到正确的Cli，将响应发送到该Cli。
不过这一复杂性，使请求-响应模式成为最强大的模式。允许将几十/几百个Cli请求均匀分布到大量的Serv中，除了设置MQ外，不需做其他任何工作。
一个优秀的MQ，允许Serv在不丢失消息的前提下，绑定到MQ/解除绑定，这种MQ同样允许Serv在需要维护而关闭时的行为，对Cli机器保持不可见。
8.3.5. req-rep的mq，是将能在某台机器上，大量运行的多个轻量级thd(如网络前端Serv得到许多线程)与DB-Cli/文件Serv连接起来的一种很好的方式。DB-Cli/文件Serv有时需要被调用，替代前端Serv进行一些高负荷的运算。
req-rep适用于RPC机制，且提供了普通RPC系统没有提供的额外优点:
许多消费者/生产者，都可使用扇入/扇出的工作模式，绑定到同一个队列，而模式的不同，对于Cli来说是不可见的。
8.3.6. 在Python中使用消息队列
8.3.6.1. 最流行的MQ被实现为独立的Serv。构建程序时，为了完成各种任务选用的所有组件(如生产者、消费者、过滤器、RPC服务)都可以绑定到MQ，且相互不知道彼此的地址，甚至不知道彼此的身份。
AMQP协议是最常见的跨语言MQ协议实现之一，可安装许多支持AMQP协议的开源Serv，如RabbitMQ、Apache QpidServ等
8.3.6.2. 很多程序员从来不学些消息协议。相反，会去依赖一些三方库，将MQ的重要功能封装起来，提供易于使用的API。
如，许多使用Django的Py程序，会使用Celery分布式任务队列，不去学习AMQP协议。这些库可支持其他后端服务，使其不依赖于特定的协议。
Celery中，可使用简单的Redis键值存储，作为MQ，不需使用专门的消息机制。
8.3.7. 为了更好阐述重点，使用不需要安装全功能，独立MQ-Serv的例子，会更方便。∅MQ(Zero Message Queue)，由提出AMQP的公司开发，将提供智能消息机制的任务，交给了每个MQ-Cli程序完成，没有交给集中式Serv处理。
只需将∅MQ库嵌入到自己的每个程序中，就能同时构建消息机制，无需使用一个集中式Serv。与基于集中式Serv的架构，在可靠性、冗余性、重传、磁盘永久存储等方面有诸多不同。
8-3示例使用简单、可能、不高效的蒙特卡洛方法计算π的值。
消息传递的拓扑结构相当重要，图8-1展示了结构：
bitsource------------>发布、订阅'00'->always_yes---------PUSH PULL->tally
/ ----------->发布、订阅'01''10''11'->judge---------->请求、响应->pythagoras
/ ---------->PUSH PULL->tally
1)bitsource程序生成一个长为2π的str，str由0和1组成。其中，使用奇数位的n为数字表示x轴坐标，使用偶数位的n位数字表示y轴坐标(坐标值均为无符号整数).
该坐标值对应的点，是否落在以坐标原点为中心，以n位整数表示的坐标值为半径的象限1/4的圈内?
2)使用发布者-订阅者结构的MQ，构建两个监听模块，对bitsource生成表示坐标轴的二进制str进行监听。
3)always_yes监听模块，只接受以00开始的str，然后直接生成结果Y，并推送给tally模块。
4)如果str的前两位均为0，那么x轴和y轴坐标一定都小于坐标最大值的一半，所以对应的点一定位于第一象限的1/4圆内
5)如果str开始为01/10/11,就必须通过judge模块处理，真正进行测试。
6)judge模块会请求pythagoras模块，计算两个int坐标值的平方和，判断对应的点是否在第一象限的1/4圆内，根据结构，将T/F推送到输出队列中。
7)最下面的tally模块，接收由每个随机串生成的T/F,通过计算T的数量与T和F总数的比值，就能对π值进行估算。
8-3实现了包含5个模块的拓扑结构，持续运行了30s，该程序需∅MQ:
# 8-3 连接5个不同模块的∅MQ消息机制 queuecrazy.py
# !/usr/bin/env python
# -- coding: utf-8 --
import random, threading, time, zmq
B = 32 # number of bits of precision in each random integer
def ones_and_zeros(digits):
"""Express 'n' in at least 'd' binary digits, with no special prefix."""
return bin(random.getrandbits(digits)).Istrip('ob').zfill(digits)
def bitsource(zcontext, url):
"""Produce random points in the unit square."""
zsock = zcontext.socket(zmq.PUB)
zsock.bind(url)
while True:
zsock.send_string(ones_and_zeros(B * 2))
time.sleep(0.01)
def always_yes(zcontext, in_url, out_url):
"""Coordinates in the lower-left quadrant are inside the unit cicle."""
isock = zcontext.socket(zmq.SUB)
isock.connect(in_url)
isock.setsockopt(zmq.SUBSCRIBE, b'00')
osock = zcontext.socket(zmq.PUSH)
osock.connect(out_url)
while True:
isock.recv_string()
osock.send_string('Y')
def judge(zcontext, in_url, pythagoras_url, out_url):
"""Determing whether each input coordinate is inside the unit circle."""
isock = zcontext.socket(zmq.SUB)
isock.connect(in_url)
for prefix in b'01', b'10', b'11':
isock.setsockopt(zmq.SUBSCRIBE, prefix)
psock = zcontext.socket(zmq.REQ)
psock.connect(pythagoras_url)
osock = zcontext.socket(zmq.PUSH)
osock.connect(out_url)
unit = 2 ** (B * 2)
while True:
bits = isock.recv_string()
n, m = int(bits[::2], 2), int(bits[1::2], 2)
psock.send_json((n, m))
sumsquares = psock.recv_json()
osock.send_string('Y' if sumsquares < unit else 'N')
def pythagoras(zcontext, url):
"""Return the sum-of-squares of number sequences."""
zsock = zcontext.socket(zmq.REP)
zsock.bind(url)
while True:
numbers = zsock.recv_json()
zsock.send_string(sum(n * n for n in numbers))
def tally(zcontext, url):
"""Tally how many points fall within the unit circle, and print pi."""
zsock = zcontext.socket(zmq.PULL)
zsock.bind(url)
p = q = 0
while True:
decision = zsock.recv_string()
q +=1
if decision == "Y":
p += 4
print(decision, p / q)
def start_thread(function, *args):
thread = threading.Thread(target=function, args=args)
thread.daemon = True # so you can easily Ctrl-C the whole program
thread.start()
def main(zcontext):
pubsub = "tcp://127.0.0.1:6700"
reqrep = "tcp://127.0.0.1:6701"
pushpull = "tcp://127.0.0.1:6702"
start_thread(bitsource, zcontext, pubsub)
start_thread(always_yes, zcontext, pubsub, pushpull)
start_thread(judge, zcontext, pubsub, reqrep, pushpull)
start_thread(pythagoras, zcontext, reqrep)
start_thread(tally, zcontext, pushpull)
time.sleep(30)
if __name__ == '__main__':
main(zmq.Context())
每个thd都创建了各自的一个/多个用于通信的socket，原因是，试图让两个线程共享一个mq-socket不安全。
多个thd间确实共享了一个共同的上下文(context)对象，能保证所有thd都存在于同一个URL和MQ空间内。只要为每个prc创建一个∅MQ上下文即可。
8.3.8. 尽管这些socket提供的方法名与recv()和send()这样的普通socket操作名类似，但有着不同的语义。消息是有序保存的，且从不重复。在连续的消息流中，消息被分隔为了多个独立的消息，且不会丢失。
此例能在不多的代码汇总使用典型消息队列提供的大多数主流消息模式。
always_yes和judge与bitsource间的连接，构成了一个pub-sub系统。
该系统中，所有连接的Cli都会接收到一份来自发布者的消息(被顾虑的消息不会发送给相应的订阅者)。每个设置了过滤器的∅MQ-socket都会接收所有的，起始两位字符与过滤str相匹配的消息。保证了两个sub能接收由bitsource生成的所有str，原因在于"00""01""10""11"这4中过滤器覆盖了str开头两位的所有可能组合。
8.3.9. judge与pythagoras间，是典型的RPC-req-rep关系。创建了REQ-socket的Cli，必须先发起请求，将消息发送给绑定到该socket的某一个等待中的代理。消息机制自动为请求，隐式添加了一个返回地址。一旦代理完成了工作，并发出了相应，就可使用这个返回地址通过REP-socket将响应发送到正确的Cli(即使有几十/百个Cli绑定到了socket，也可正确工作)
8.3.10. tally工作进程阐释了推送-拉取(push-pull)模式的工作原理。能保证被推送的每一项，都会被一个且仅被一个连接到该socket的代理接收。如果启动了多个tally-prc，么个来自上游的数据都只会发送给其中的一个tally-prc，会分别计算π值。
∅MQ不需考虑bind()和connect()的调用顺序。如果某个通过URL描述的终端后才会启动，那么∅MQ会通过延时设置是和轮询，不断地隐式重试connect()。即使在程序运行过程中，某个代理掉线，∅MQ也能保持健壮性
$ python queueapi.py
...
Y 3.14...
需要保证消息的传输，在消息无法被处理时，需要将它们持久化保存，此外，还需进行一些流量控制，保证代理在速度较慢时，也能处理队列中处于等待状态的消息的数量。
通常需要使用一些更为复杂的模式。使用RabbitMQ、Qpid和Celery底层的Redis的全功能MQ时，只需做很少量的工作，却能保证极低的出错率
8.4. 小结
8.4.1. Memcached利用所有安装了它的Serv上的空闲RAM构建了一个大型的LRU缓存，只要程序需要删除/替换过时的记录/要处理的数据会在一段固定且可预测的时间内过期，Memcached就可大大减少DB/其他后端存储的运行负荷。可在处理过程中的多个不同地方插入Memcached。
如，保存一个花销很大的DB查询结果，不如直接将最终生成的网络图形界面元素，存入缓存。