python运维之路——Twsited、memcache、Redis、RabbitMQ

Twsited网络框架

Twisted是一个事件驱动的网络框架，其中包含了诸多功能，例如：网络协议、线程、数据库管理、网络操作、电子邮件等等

事件驱动

简而言之，事件驱动分为两个部分，第一、注册事件；第二、出发事件。

自定义事件驱动框架

#!/usr/bin/env python3
#自定义事件驱动框架
event_list = []

def run():
for event in event_list:
obj = event()
obj.execute()

class BeasHandler(object):
#用户必须集继承该类。从而规范所有类的方法（类似于接口的功能）
def execute(self):
raise Exception('你必须自己创建该方法')

#!/usr/bin/env python3
#用户使用自定义的事件驱动框架

import Twsited_test01
class MyHandler(Twsited_test01.BeasHandler):
def execute(self):
print('自定义的事件驱动——MyHandler')

Twsited_test01.event_list.append(MyHandler)
Twsited_test01.run()

protocols

Protocol买描述了如何以异步的方式处理网络中的事件。HTTP、DNS以及IMAP是应用层协议中例子。

Protocols实现了IProtocols接口，它包含如下的方法:

- makeConnection 在transport对象和服务器之间建立一条连接

- connectionMade 链接建立起来后调用

- dataReceived 接收数据时调用

- connectionLost 关闭链接时调用

Transport

Transport代表网络中两个通信节点之间的连接。Transport负责描述连接的细节，比如连接是面向流式的还是面向数据包的，流控以及可靠性。TCP、UDP和Unix套接字可作为transport的例子。它们被设计为“满足最小功能单元，同时具有最大程度的可复用性“，而且从协议实现中分离出来，这让许多协议可以采用相同类型的传输。Transport实现了ITransport接口，它包含如下方法：

- write 以非阻塞的方式按顺序依次将数据写到物理连接上

- writeSequence 将一个字符串列表写到物理连接上

- loseConnection 将所有挂起的数据写入，然后关闭连接

- getPeer 取得连接中对端的地址信息

- getHost 取得连接中本端的地址信息

将transport从协议中分离出来也是的这两个层次的测试变得更加简单。可以通过简单地写入一个字符串来模拟传输，用这种方式来检查。

一个学说话服务器的例子

#!/usr/bin/env python3
#server端

from twisted.internet import protocol
from twisted.internet import reactor
#reactor实现了对端口的监听

#用户自己定义了一个类
class Echo(protocol.Protocol):
#框架的内部方法，再接收数据的时候自动调用
def dataReceived(self, data):
#已非阻塞的形式发送数据
self.transport.write(data)

def main():
#将server端实例化（按照框架的要求）
factory = protocol.ServerFactory()
#运行的时候会自动调用protocol，也就是自动电泳Echo
factory.protocol = Echo
#监听
reactor.listenTCP(12345, factory)
reactor.run()

if __name__ == '__main__':
main()

#!/usr/bin/env python3
#client端

from twisted.internet import reactor, protocol

#自己定义的连接类，一次性链接，发送，接收，显示
class EchoClient(protocol.Protocol):
#连接建立起来后自动调用
def connectionMade(self):
self.transport.write(bytes("hello dingyi", encoding='utf8'))

#收到数据的时候自动调用
def dataReceived(self, data):
print("服务端返回的信息：%s" % str(data, encoding = 'utf8'))
#主动断开链接
self.transport.loseConnection()

#链接丢失的时候自动调用
def connectionLost(self, reason):
print('链接断开')

#自己定义的客户端类
class EchoFactory(protocol.ClientFactory):
#将自己定义好的连接类赋值（按照框架的要求）
protocol = EchoClient

#当客户端启动失败的时候调用
def clientConnectionFailed(self, connect
1b5d8
or, reason):
print('连接失败')
reactor.stop()
#当客户端链接断开的时候调用
def clientConnectionLost(self, connector, reason):
print('链接丢失')
reactor.stop()

def main():
#实例化
f = EchoFactory()
#写localhost会失败
reactor.connectTCP("127.0.0.1", 12345, f)
reactor.run()

if __name__ == '__main__':
main()

运行服务器脚本将会启动一个TCP服务器，监听端口12345上的连接。服务器采用的是Echo协议，数据经TCP transport对象写出。运行客户端脚本对服务器发起一个TCP连接，回显服务端的回应后终止连接并停止reactor事件循环。这的Factory用来对连接的双方生成protocol对象实例。两端的通信是异步的，connectTCP负责注册回调函数到reactor事件循环中，当socket上有数据可读时通知回调处理。

一个传送文件的例子

#!/usr/bin/env python3
#server端
import optparse, os
#optparse 负责处理脚本后面的参数
from twisted.internet.protocol import ServerFactory, Protocol

def parse_args():
#脚本的ReadMe
usage = '''usage: %prog [options] poetry-file

This is the Fast Poetry Server, Twisted edition.
Run it like this:

python fastpoetry.py <path-to-poetry-file>

If you are in the base directory of the twisted-intro package,
you could run it like this:

python twisted-server-1/fastpoetry.py poetry/ecstasy.txt

to serve up John Donne's Ecstasy, which I know you want to do.
'''
#添加readme
parser = optparse.OptionParser(usage)
#某个选项的help信息
help = 'The port to listen on. Default to a random available port.'
#添加选项、后面参数的类型、以及help信息
parser.add_option('--port', type='int', help = help)

help = "The interface to listen on. Default is localhost."
parser.add_option('--iface', help=help, default='localhost')

#获取选项和参数,options是一个字典，选项是key，选项的值是value。args是个列表，里面是剩下的参数
#如果参数是--por 或者--p、--po等，不会报错，如果是--porta，就会报错，信息是usage的内容
options, args = parser.parse_args()
print('--args:',options, args)

if len(args) != 1:
parser.error('只能传送一个文件')
#需要传输的文件
poetry_file = args[0]

if not os.path.exists(args[0]):
parser.error("找不到这个文件%s" % poetry_file)

return options, poetry_file

#执行的动作的类，用户自己定义，但是必须继承Protocol
class PoetryProtocol(Protocol):
#当连接建立起来的时候调用
def connectionMade(self):
#将文件内容写入物理连接，在执行的动作的类中，想要执行上一层定义的方法或属性，必须通过factory来调用
self.transport.write(bytes(self.factory.poem, encoding = 'utf8'))
#断开连接
self.transport.loseConnection()

#按框架的要求继承指定的类，定义服务端类
class PoetryFactory(ServerFactory):
#定义执行的动作的类，必须用protocol进行实例化
protocol = PoetryProtocol
def __init__(self, poem):
self.poem = poem

def main():
#获取选项和需要传输的文件
options, poetry_file = parse_args()
#读取需要传输的文件
poem = open(poetry_file).read()
#实例化（传送文件用）
factory = PoetryFactory(poem)
from twisted.internet import reactor
#建立连接，监听默认9000端口，IP用interface
port = reactor.listenTCP(options.port or 9000, factory, interface = options.iface)
#打印 本机地址和文件名
print('serving %s on %s.' % (poetry_file, port.getHost()))
#运行reactor.listenTCP中的factory，也就是PoetryFactory(poem)中的protocol，也就是PoetryProtocol
reactor.run()

if __name__ == '__main__':
main()

#!/usr/bin/env python3
#客户端

import optparse
from twisted.internet.protocol import Protocol, ClientFactory

def parse_args():
usage = '''usage: %prog [options] [hostname]:port ...
This is the Get Poetry Now! client, Twisted version 3.0
Run it like this:
python get-poetry-1.py port1 port2 port3 ...
'''
parser = optparse.OptionParser(usage)
#第一个变量是字典，这里不需要，使用下划线表示这个变量是多余的
_,addresses = parser.parse_args()
#如果脚本没加参数，打印帮助文档
if not addresses:
print(parser.format_help())
parser.exit()

def parse_address(addr):
if ':' not in addr:
host = '127.0.0.1'
port = addr
else:
#确保分离后是两个变量
host, port = addr.split(':', 1)
#判定最后的端口是不是只有数字
if not port.isdigit():
parser.error('端口必须是数字')

return host, int(port)
#addresses是一个列表，将列表中的每个元素，当做参数传入pares_address中，并把所有结果组合成列表返回
return map(parse_address, addresses)

#自定义的执行动作的类，框架要求必须继承Protocol
class PoetryProtocol(Protocol):
poem = ''

#接收数据的时候调用
def dataReceived(self, data):
data = str(data, encoding='utf8')
self.poem += data
#连接断开的时候调用
def connectionLost(self, reason):
self.poemReceived(self.poem)

def poemReceived(self, poem):
#想要调用外部类的方法，需要通过factory,在这里实际执行的是外部类的callback方法
self.factory.poem_finished(poem)

#用户自己定义的客户端类，框架规定必须继承ClientFactory
class PoetryClientFactory(ClientFactory):
#安框架的要求，把定义的执行动作的类赋值给protocol
protocol = PoetryProtocol
#初始化callback，用来接收callback函数
def __init__(self, callback):
self.callback = callback
#调用callback函数
def poem_finished(self, poem):
self.callback(poem)

#启动监听
def get_poetry(host, port, callback):
"""
Download a poem from the given host and port and invoke callback(poem)
when the poem is complete.
"""

from twisted.internet import reactor
#监听端口
factory = PoetryClientFactory(callback)
reactor.connectTCP(host, port, factory)

def poetry_main():
#获取地址和端口
addresses = []
get_map = parse_args()
for i in get_map:
addresses.append(i)
from twisted.internet import reactor
#用来存储多个服务端发来的文件,脚本只调用了poetry_main这一个函数，所以可以把poems当做全局变量
poems = []

#这个函数是检测poems的长度和addresses的长度，由于这里是主函数，脚本中的所有函数都能读取到这两个变量
def got_poem(poem):
poems.append(poem)
#当收到的文件个数跟用户填写的地址个数一样的时候，停止
if len(poems) == len(addresses):
reactor.stop()
#遍历地址,按地址数量起监听
for address in addresses:
host, port = address
get_poetry(host, port, got_poem)
#运行，等待连接
reactor.run()
#打印文件内容
for poem in poems:
print(poem)

if __name__ == '__main__':
poetry_main()

Twsited深入

twsited很深奥，需要很长事件的学习，这里只是介绍了部分的内容，更具体的部分请参看下面的博客

http://krondo.com/an-introduction-to-asynchronous-programming-and-twisted/

http://blog.csdn.net/hanhuili/article/details/9389433

缓存数据库介绍

NoSQL(NoSQL = Not Only SQL)，即“不仅仅只是SQL”，泛指非关系行的数据库，随着互联网web2.0网站的兴起，传统的关系型数据库在应付web2.0网站，特别是超大规模和高并发的SNS类型的web2.0存动态网站已经显得力不从心了，暴露了很多难以克服的问题，而非关系型的数据库由于其本身的特点，得到了非常迅速的发展。NoSQL数据库的产生就是为了解决大规模数据集合多种数据种类带来的挑战，尤其是大数据应用难题。

NoSQL数据库的四大分类

键值(key-value)存储数据库

这一类数据库主要会使用到一个哈希表，这个表中有一个特定的键和一个指针指向特定的数据。Key/value模型对于IT系统来说有优势在于简单，易部署。但是DBA支队部分值进行查询或更新的时候，Key-Value就显得效率低下了。例如：Tokyo Cabinet/Tyrant, Redis, Voldemort, Oracle BDB.

典型应用场景：内容缓存，主要用于处理大量数据的高访问负载，也用于一些日志系统等。

数据模型：Key指向Value的键值对，通常是hash table来实现

优点：查找速度快

缺点：数据无结构化，通常指被当做字符串或者二进制数据

列存储数据库

这部分数据通常是用来应对分布式存储的海量数据。键仍然存在，但是他的特点是指向了多个列。这些列是由列家族安排的。如：Cassandra, HBase, Riak.

典型应用场景：分布式文件系统

数据模型：以列簇式存储，将同一列数据存在一起

优点：查找速度快，可扩展性强，更容易进行分布式扩展

缺点：功能相对局限

文档型数据库

文档型书库的灵感来自于Lotus Notes办公软件，而且他同第一种键值存储类似。该模型的数据模型是版本化的文档，半结构化的文档以特定的格式存储，比如JSON。文档型数据库可以看做是键值型数据库的升级版，允许之间嵌套键值。而且文档型数据库比键值型数据的查询效率更高。如：CouchDB, MongoDb. 国内也有文档型数据库SequoiaDB，已经开源。

典型应用场景：Web应用（与Key-Value类似，Value是结构化的，不同的是数据库能够了解Value的内容）

数据模型：Key-Value对应的键值对，Value为结构化数据

优点：数据结构要求不严格，表结构可变，不需要像关系型数据库那样需要预先定义表结构

缺点：查询性能不高，而且缺乏统一的查询语法

图形(Graph)数据库

图形结构的数据库同其他行列以及刚性结构的SQL数据库不同，它是使用灵活的图形模型，并且能够扩展到多个服务器上。NoSQL数据库没有标准的查询语言(SQL)，因此进行数据库查询需要定制数据模型。许多NoSQL数据库都有REST式的数据接口或者查询API。例如：Neo4J, InfoGrid, Infinite Graph.

典型应用场景：社交网络、推荐系统等。专注于构建关系图谱

数据模型：图结构

优点：利用图结构相关算法，比如最短路径寻址，N度关系查找等

缺点：很多时候需要对整个图做计算才能得出需要的信息，而且这种结构不好做分布式的集群方案。

总结

NoSQL数据库在以下情况下比较试用：

1. 数据模型比较简单

2. 需要灵活性更强的IT系统

3. 对数据库性能要求高

4. 不需要高度的数据一致性

5. 对给定Key，比较容易映射复杂值的环境

memcached

Memcached 是一个高性能的分布式内存对象缓存系统，用于动态Web应用以减轻数据库负载。他通过在内存中缓存数据和对象来减少读取数据库的次数，从而提高动态、数据库驱动网站的速度。Memcached基于一个存储键/值对的hashmap。其守护进程(daemon)是用C写的，但是客户端可以用任何语言来编写，并通过memcached协议与守护进程通信。

安装memcached

sudo apt-get install memcached

启动Memcached

memcached -d -m 10 -u root -l 127.0.0.1 -p 11100 -c 128 -P /tmp/memcached.pid
参数说明：
-d 是启动一个守护进程
-m 是分配给Memcache使用的内存数量，单位是MB
-u 是运行mencached的用户
-l 是监听的服务器IP地址
-p 是设置Memcached监听的端口，最好是1024以上的端口
-c 是最大运行的并发连接数，默认是1024，按照尼福取其的负载量来设定
-P 是设置保存Memcached的pid文件

memcached命令

- 存储命令：set,add,replace,append,prepend,cas

- 获取命令：get，gets

- 其他命令：delete、stats..

python操作memcached

安装API

sudo pip3 install Python-memcached

1、第一个小例子

#!/usr/bin/env python3
import memcache
#debug=True 表示运行出现错误时，显示错误信息，在生产环境中移除该参数
mc = memcache.Client(['127.0.0.1:11100',], debug=True)
mc.set('foo', 'bar')
ret = mc.get('foo')
print(ret)

py@py-dy:~/桌面/memcached$ python3 memcached_test01.py
bar

2、自带集群功能

Python-memcached模块原声支持集群操作，其原理实在内存维护一个主机列表，且集群中主机的权重值和主机在列表中重复出现的次数成正比

主机    权重
1.1.1.1   1
1.1.1.2   2
1.1.1.3   1

那么在内存中主机列表为：
host_list = ["1.1.1.1", "1.1.1.2", "1.1.1.2", "1.1.1.3", ]

如果用户根据需要在内存中创建一个键值对（如：k1 = ‘v1’），那么要执行下一步骤：

- 根据算法价将K1转化成一个数字

- 将数字和主机列表长度求余数，得到一个值N(0<=N<=列表长度)

- 在主机列表中根据第2步得到的值为索引获取主机，例如：host_list

- 连接 将第3步中获取的主机，将k1=’v1’放置在该服务器的内存中

代码实现如下：

mc = memcache。Client([('192.168.0.1:12000',1), ('192.168.0.2:12000', 2), ('192.168.0.3:12000',1)], debug = True)
mc.set('k1':'v1')

3、add

添加一条键值对，如果key存在则报错

#!/usr/bin/env python3
import memcache

mc = memcache.Client(['127.0.0.1:11100'], debug = True)
mc.add('k2','v2')
#第二次执行时报的错
#MemCached: while expecting 'STORED', got unexpected response 'NOT_STORED'

4、relpace

修改某个KEY的值，如果key不存在，则异常

#!/usr/bin/env python3
import memcache
mc = memcache.Client(['127.0.0.1:11100'], debug = True)
mc.replace('k2','v11')
#报错的异常
#MemCached: while expecting 'STORED', got unexpected response 'NOT_STORED'

5、set和set_multi

set 设置一个键值对，如果key不存在，则创建，如果key存在则修改

set_multi 设置对个键值对，如果key不存在，则创建，如果key存在则修改

#!/usr/bin/env python3
import memcache

mc = memcache.Client(['127.0.0.1:11100'], debug = True)
mc.set('k4','v4')
mc.set_multi({'k01':'v01', 'k02':'v02'})

delete和delete_mnlti

delete 在memcached中删除一个指定的键值对

delete_multi 在memcached中删除多个指定的键值对

#!/usr/bin/env python3
import memcache

mc = memcache.Client(['127.0.0.1:11100'], debug = True)
mc.delete('k4')
mc.delete_multi(['k01', 'k02'])

get和get_multi

get 获取一个键值对

get_multi 获取多个键值对

#!/usr/bin/env python3
import memcache
mc = memcache.Client(['127.0.0.1:11100'], debug=True)
val = mc.get('k2')
print('val:', val)

vals = mc.get_multi(['k1', 'k2'])
print('vals:', vals)

append和prepend

append 修改指定key的值，在该值后面追加内容

prepend 修改指定key的值，在该值前面插入内容

#!/usr/bin/env python3
import memcache

mc = memcache.Client(['127.0.0.1:11100'], debug = True)

mc.add('k1', 'v1')
mc.append('k1', 'after')
mc.prepend('k1', 'before')

a = mc.get('k1')
print(a)

decr和incr

incr 自增，将memcached中的某一个值增加N(N默认为1)

decr 自减，将memcached中的某一个值减少N(N默认为1)

#!/usr/bin/env python3
import memcache

mc = memcache.Client(['127.0.0.1:11100'], debug= True)
mc.set('k1', '777')

mc.incr('k1')
#k1 = 778
mc.incr('k1', 10)
#k1 = 788

mc.decr('k1')
#k1 = 787
mc.decr('k1', 10)
#k1 = 777

gets和cas

类比商城商品剩余个数，假设改值保存在memcache中，product_count = 900

A用户刷新页面从memcache中读取到product_count = 900

B用户刷新页面从memcache中读取到product_count = 900

如果A、B均购买商品

A用户修改商品剩余个数，product_count = 899

B用户修改商品剩余个数，product_count = 899

如此一来缓存内的数据就不正确了，两个用户购买商品后剩余个数是899

如果使用python的set和get来操作上述过程，就会得到上述结果

如果想要避免这种情况，就需要使用gets和cas

#!/usr/bin/env python3
import memcache
mc = memcache.Client(['127.0.0.1:11100'], debug = True)
mc.set('product_count', '900')
v = mc.gets('product_count')
print(v)
#如果有人在gets之后，cas之前修改了ptoduct_count，那么，下面的设置将会执行失败，引发异常，从而避免非正确数据的生成
mc.cas('product_count', '889')

v = mc.get('product_count')
print(v)

注：本质上每次执行gets时，会从memcache中获取一个自增数字，通过cas去修改gets的值时，会携带之前获取的自增值和memcache中的自增值进行比较，如果相等，则可以提交，如果不相等，那表示在gets和cas执行之间，又有其他人修改了gets(获取了缓冲的指定值)，这就有可能产生不正确的数据，所以就不能修改该值了

Redis

介绍

redis是业界主流的key-vlaue nosql数据库之一。和memcached类似，它支持存储的v安略类型相对更多，保罗string(字符)，list(列表)，set(集合)，zset(有序集合)和hash(哈希类型)，这些数据类型都支持push/pop、add/remove及取交集和差集以及更丰富的操作，而且这些操作都是原子性的（要么成功要么失败，不会只完成一半）。在此基础上，redis支持各种不同的排序。与memcached一样，为了保证效率，数据都是缓存在内存中。区别是redis会周期性的吧更新的数据写入磁盘或者把修改操作写入追加的记录文件，并在此基础上实现了主从同步。

redis优点

- 异常快速：Redis是非常快的，每秒可执行大约110000次设置操作，81000个/每秒的读取操作。

- 支持丰富的数据类型：Redis支持最大多数开发人员已经知道的如列表、集合、可排序集和、哈希等数据类型。这使得在应用中很容易解决各种问题。

- 操作都是原子的：所有Redis的操作都是原子的，从而确保当两个客户同事访问Redis服务器得到的是更新后的值（最新值）。

- MultiUtility工具：Redis是一个多功能实用工具，可以在很多如：缓存、消息传递队列中使用(Redis原生支持发布/订阅)。在应用程序中，如：Web应用程序会话，网站页面点击数等任何短暂的数据。

Redis安装环境

安装

sudo apt-get update
sudo apt-get install redis-server

启动

redis-server

查看/进入redis

py@py-dy:~/桌面/memcached$ redis-cli
127.0.0.1:6379>
#在上面的提示信息中：127.0.0.1 是本机的IP地址，6379是 Redis 服务器运行的端口。现在输入 PING 命令如果返回的是PONG就说明安装成功

python操作Redis

安装

sudo pip3 install redis

图形管理工具

要在Ubuntu 上安装 Redis桌面管理，可以从 http://redisdesktop.com/download 下载包并安装它。
Redis 桌面管理器会给你用户界面来管理 Redis 键和数据。

Redis API使用

redis-py的API使用可以分类为：

- 连接方式

- 连接池

- 操作

- string操作

- Hash操作

- List操作

- Set操作

- Sort Set操作

- 管道

- 发布订阅

连接方式

1、操作模式

redis-py提供两类Redis和StrictRedis用于实现Redis的命令，StrictRedis用于实现大部分官方的命令，并使用官方的语法和命令，Redis是StrictRedis的子类，用于向后兼容旧版本的redis-py

#!/usr/bin/env python3
import redis
r = redis.Redis(host='127.0.0.1', port=6379)
r.set('foo', 'bar')
print(r.get('foo'))
py@py-dy:~/桌面/redis$ python3 redis_test01.py
b'bar'

2、连接池

redis-py 使用connectoin poll来管理，对一个redis server的所有连接，避免每次建立、释放链接的开销。默认，每次Redis实力都会维护自己的一个连接池。可以直接建立一个连接池，然后作为参数Redis，这样就可以实现多个Redis实力共享一个连接池。

import redis

pool = redis.ConnectionPool(host='10.211.55.4', port=6379)

r = redis.Redis(connection_pool=pool)
r.set('foo', 'Bar')
print r.get('foo')

操作

1、string操作

在redis中的string在内存中按照一个name对应一个value来存储。

name value

n1 ————> v1

n2 ————> v2

n3 ————> v3

set(name, value, ex=None, px=None, nx=False, xx=False)

在redis中设置值，默认，不存在则创建、存在则修改
参数：
ex，过期时间（秒）
px，过期时间（毫秒）
nx，如果设置为True，则只有name不存在时，当前set操作才执行
xx，如果设置为True，则只有name存在时，当前set操作才执行

setnx(name, value)

设置值，只有name不存在时，执行设置操作（添加）

setex(name, value, time)

设置值，time参数是过期时间（数字秒或timedelta对象）

psetsx(name, time_ms, value)

设置值，参数time_ms是 过期时间（数字毫秒或timedelta对象）

mset(*args, **kwargs)

批量设置值
mset(k1 = 'v1', k2 = 'v2')
或
mset({'k1':'v1', 'k2':'v2'})

get(name)

获取值

mget(keys, *args)

批量获取
mget('k1', 'k2')
mget(['k1', 'k2'])

getset(name, value)

设置新值并获取原来的值

getrange(key, start, end)

获取子序列（根据字节获取，非字符）
参数：
name，Redis 的name
start，其实位置(字节)
end， 结束位置(字节)
如：'奥术大师'， 0-3表示'奥'

setrange(name, offset, value)

修改字符串内容，从指定的字符串索引开始向后替换（新值太长是则向后添加）
参数：
offset：字符串的索引，字节(一个汉子三个字节)
vlaue：要设的值

setbit(name, offset, vlaue)

对name对应值的二进制表示的位进行操作

参数：
name，redis的name
offset，位的索引(将值变换成二进制后在进行索引)
value，值只能是0或1

注：如果在redis中有一个对应：n1 = 'foo'
那么字符串foo的二进制表示为：01100110 01101111 01101111
所以，如果执行setbit('n1', 7, 1),就会将第七位设置为1（从0开始），变成01100111 01101111 01101111，即'goo'

扩展，转换二进制表示:
source = 'foo'

for i in source:
num = ord(i)
print(bin(num),replace('b',''))
如果source是汉字怎么办
对于utf-8，每一个汉字占 3 个字节
对于汉字，for循环时候会按照 字节 迭代，那么在迭代时，将每一个字节转换 十进制数，然后再将十进制数转换成二进制

用途举例：用最省空间的方式，存储在线用户以分别是那些用户在线

getbit(name, offset)

获取name对应的值的二进制便是中的某位的值(0或1)

bitcount(key, start=None, end=None)

获取name对应的值的二进制表示中1的个数
参数：
key, Redis的name
start，起始位置
end，结束位置

strlen(name)

返回name对应值的字节长度（汉字为三个字节）

incr(self, name, amount=1)

自增name对应的值，当name不存在是，则创建name = amount，否则自增
参数：
name，Redis对应的那么
amount，自增数（必须是整数）
注：同incrby

incrbyfloat(self, name, amount=1.0)

自增name对应的值，当name不存在是，则创建name = amount，否则自增
参数：
name，Redis对应的那么
amount，自增数（浮点型）

decr(self, name, amount=1)

自减name对应的值，当name不存在时，则创建name = amount，否则自减
参数：
name，Redis的name
amount，自减数(整数)

append(key, value)

在redis 那么队形的之后面追加内容
参数：
key，redis的name
value， 要追加的字符串

2、Hash操作

hash表现形式上有些像python中的dict，可以存储一组关联性较强的数据。在redis中Hash在内存中的存储格式如下图



hset(name, key, value)

name对应的hash中设置一个键值对(不存在，则创建；否则，修改)
参数：
name，redis的name
key，name对应的hash中的key
value，name对应的hash的value
注：
hsetnx(name, key, value),当那么队形的hash中不存在当前key是则创建（相当于添加）

hmset(name, mapping)

在name对应的hash中批量设置键值对
参数：
name：redis的name
mapping，字典，如：{"k1":"v1", "k2":"v2"}

如：
r.hmset('xx', {'k1':'v1', "k2":'v2'})

hget(name, key)

在neme中对应的hash中获取根据key获取的value

hmget(name, key, *args)

在那么对应的hash中获取多个key的值
参数：
name，redis对应的name
keys，要获取key的集合。如['k1', 'k2', 'k3']
*args,要获取的key，如：k1, k2, k3

如：
r.mget('xx', ['k1', 'k2'])
或
print(r.hmget('xx','k1','k2'))

hgetall(name)

获取name对应hash的所有键值

hlen(name)

获取name对应的hash中键值对的个数

hkeys(name)

获取name对应的所有的key的值

hvals(home)

获取name对应的hash中的所有value的值

hexiste(name, key)

检查name对应的hash中是否存在当前传入的key

hdle(name, *key)

将name对应的hash中指定key的键值对删除

hincrby(name, key, amount=1)

自增name对应的hash中指定key的值，不存在则创建key=amount
参数：
name, redis中的那么
key, hash对应的key
amount, 自增数(整数)

hincrbyfloat(name, key, amount=1.0)

自增name对应的hash中指定的key的值，不存在则创建key=amount
参数：
name，redis中对应的name
key，hash对应的key
amount， 自增数(浮点数)

自增name对应的hash中指定key的值，不存在则创建key=amount

hscan(name, cursor=0, match=None, count=None)

自增式迭代获取，对于数据大的数据非常有用。hscan可以实现分片的获取数据，并非一次性将数据全部取完，从而导致内存溢出
参数：
name，redis的name
cursor，游标（基于游标分批获取数据）
match，匹配指定key。默认None表示所有key
count，每次分片最少获取个数，默认None表示采用Redis的默认分片个数

例：
第一次：cursor1, data1 = r.hscan('xx', cursor=0, match=None, count=None)
第二次：cursor2, data1 = r.hscan('xx'，cursor=cursor1, match=None, count=None)
...
直到返回值cursor的值为0时，表示数据已经通过分片获取完毕

hscan_iter(name, match=None, count=None)

利用yield封装hash创建生成器，实现分批去redis中获取数据
参数：
match，匹配指定key，默认None表示所有的key
count，每次分片最少获取个数，默认None表示采用Redis的默认分片个数

例：
for item in r.hscan_iter('XX'):
print(item)

3、list

List操作，redis中的list在内存中按照一个name对应一个List来存储。

name list

l1 ————> [v1,v2,v3…]

l2 ————> [v4,v5,v6…]

lpush(name, values)

在name对应的list中添加元素，每个新的元素都添加到列表的最左边
例：
r.lpush('oo', 11,22,33)
保存的顺序位:33,22,11

扩展：
rpush(name, values) 表示从右向左操作

lpushx(name, value)

在name对应的list中添加元素，只有name已经存在时，值添加到列表的最左端
扩展：
rpushx(name, value)表示从右向左操作

llen(name)

name对应的list元素个数

linsert(name, where, refvalue, value)

在name对应的列表的某一个值前或后插入一个新值
参数：
name,redis的name
where, BEFORE或AFTER
refvalue，标杆值（列表中元素的值）,在他前后插入
value，要插入的数据

lset(name, index, value)

对name对应的list中的某一个索引位置重新赋值
参数：
name,reedis的name
index,List的索引位置
value，要设置的值

lrem(name, value, num)

在name对应的list中删除指定的值
参数：
name，redis的name
value，要删除的值
num，num=0，删除 列表中所有的指定值
num=2, 从前到后删除两个指定值
num=-2，从后到前删除两个指定值

lpop(name)

在name对应的列表的左侧获取第一个元素并在列表中移除，返回值是第一个元素

扩展：
rpop(name) 表示从右向左操作

lindex(name, index)

在name对应的列表中根据索引获取列表元素

lrange(name, start, end)

在name对应的列表分片获取数据
参数：
name,redis的name
start，索引的起始位置
end，索引结束的位置

ltrim(name, start, end)

在name对应的列表中移除没有在start-end索引之间的值
参数：
name，redis的name
start，索引的起始位置
end，索引结束位置

rpoplpush(src, dst)

从一个列表取出最右边的元素，同时将其添加至另一个列表的最左边
参数：
src，要取数据的列表的name
dst，要添加数据的列表的name

blpop(keys, timeout)

将多个列表排列，按照从左到右去pop对应列表的元素

参数：
keys，redis的name的集合
timeout，超时时间，当元素所有列表的元素获取完之后，阻塞等待列表内有数据的时间（秒）, 0 表示永远阻塞
例子：
>>> r.lpush('gg',44,33,22,11)
4
>>> r.lpush('hh',44,33,22,11)
4
>>> r.blpop(('gg','hh'))
(b'gg', b'11')
>>> r.blpop(('gg','hh'))
(b'gg', b'22')
>>> r.blpop(('gg','hh'))
(b'gg', b'33')
>>> r.blpop(('gg','hh'))
(b'gg', b'44')
>>> r.blpop(('gg','hh'))
(b'hh', b'11')
>>> r.blpop(('gg','hh'))
(b'hh', b'22')
>>> r.blpop(('gg','hh'))
(b'hh', b'33')
>>> r.blpop(('gg','hh'))
(b'hh', b'44')
>>> r.blpop(('gg','hh'))
更多：
r.brpop(keys, timeout)，从右向左获取数据

brpoplpush(src, dst, timeout=0)

从一个列表的右侧移除一个元素并将其添加到另一个列表的左侧

参数：
src，取出并要移除元素的列表对应的name
dst，要插入元素的列表对应的name
timeout，当src对应的列表中没有数据时，阻塞等待其有数据的超时时间（秒），0 表示永远阻塞

自定义增量迭代

由于redis类库中没有提供对列表元素的增量迭代，如果想要循环name对应的列表的所有元素，那么就需要：
1、获取name对应的所有列表
2、循环列表
但是，如果列表非常大，那么就有可能在第一步时就将程序的内容撑爆，所有有必要自定义一个增量迭代的功能：

def list_iter(name):
"""
自定义redis列表增量迭代
:param name: redis中的name，即：迭代name对应的列表
:return: yield 返回 列表元素
"""
list_count = r.llen(name)
for index in xrange(list_count):
yield r.lindex(name, index)

使用
for item in list_iter('pp'):
print item

4、set集合操作

sadd(name,values)

name对应的集合中添加元素

scard(name)

获取name对应的集合中元素个数

sdiff(keys, *args)

在第一个name对应的集合中且不在其他name对应的集合的元素集合

sdiffstore(dest, keys, *args)

获取第一个name对应的集合中且不在其他name对应的集合，再将其新加入到dest对应的集合中

sinter(keys, *args)

获取第一个name对应集合的并集

sinterstore(dest, keys, *args)

获取多一个name对应集合的并集，再讲其加入到dest对应的集合中

sismember(name, value)

检查value是否是name对应的集合的成员

smembers(name)

获取name对应的集合的所有成员

smove(src, dst, value)

将某个成员从一个集合中移动到另外一个集合

spop(name)

从集合的右侧（尾部）移除一个成员，并将其返回

srandmember(name, numbers)

从name对应的集合中随机获取 numbers 个元素

srem(name, values)

在name对应的集合中删除某些值

sunion(keys, *args)

获取多一个name对应的集合的并集

sunionstore(dest,keys, *args)

获取多一个name对应的集合的并集，并将结果保存到dest对应的集合中

sscan(name, cursor=0, match=None, count=None)

sscan_iter(name, match=None, count=None)

同字符串的操作，用于增量迭代分批获取元素，避免内存消耗太大

5、有序集合

有序集合，在集合的基础上，为每元素排序；元素的排序需要根据另外一个值来进行比较，所以，对于有序集合，每一个元素有两个值，即：值和分数，分数专门用来做排序。

zadd(name, *args, **kwargs)

在name对应的有序集合中添加元素
如：
zadd('zz', 'n1', 1, 'n2', 2)
或
zadd('zz', n1=11, n2=22)

zcard(name)

获取name对应的有序集合元素的数量

zcount(name, min, max)

获取name对应的有序集合中分数 在 [min,max] 之间的个数

zincrby(name, value, amount)

自增name对应的有序集合的value对应的分数

r.zrange( name, start, end, desc=False, withscores=False, score_cast_func=float)

按照索引范围获取name对应的有序集合的元素

参数：
name，redis的name
start，有序集合索引起始位置,从0开始（非分数）
end，有序集合索引结束位置（非分数）
desc，排序规则，默认按照分数从小到大排序
withscores，是否获取元素的分数，默认只获取元素的值
score_cast_func，对分数进行数据转换的函数

更多：
从大到小排序
zrevrange(name, start, end, withscores=False, score_cast_func=float)

按照分数范围获取name对应的有序集合的元素
zrangebyscore(name, min, max, start=None, num=None, withscores=False, score_cast_func=float)
从大到小排序
zrevrangebyscore(name, max, min, start=None, num=None, withscores=False, score_cast_func=float)

zrank(name, value)

获取某个值在 name对应的有序集合中的排行（从 0 开始）
更多：
zrevrank(name, value)，从大到小排序

zrem(name, values)

删除name对应的有序集合中值是values的成员
如：zrem('zz', ['s1', 's2'])

zremrangebyrank(name, min, max)

根据排行范围删除

zremrangebyscore(name, min, max)

根据分数范围删除

zscore(name, value)

获取name对应有序集合中 value 对应的分数

zinterstore(dest, keys, aggregate=None)

获取两个有序集合的交集，如果遇到相同值不同分数，则按照aggregate进行操作
aggregate的值为:  SUM  MIN  MAX

zunionstore(dest, keys, aggregate=None)

获取两个有序集合的并集，如果遇到相同值不同分数，则按照aggregate进行操作
aggregate的值为:  SUM  MIN  MAX

zscan(name, cursor=0, match=None, count=None, score_cast_func=float)

zscan_iter(name, match=None, count=None,score_cast_func=float)

同字符串相似，相较于字符串新增score_cast_func，用来对分数进行操作

6、其他操作

delete(*names)

删除redis中的任意数据类型

exists

检测redis的name是否存在

keys(pattern=’*’)

根据模型获取redis的name

更多：
KEYS * 匹配数据库中所有 key 。
KEYS h?llo 匹配 hello ， hallo 和 hxllo 等。
KEYS h*llo 匹配 hllo 和 heeeeello 等。
KEYS h[ae]llo 匹配 hello 和 hallo ，但不匹配 hillo

expire(name, time)

为一个redis的某个name设置超时时间

rename(src, dst)

对redis的name重命名为

move(name, db)

将redis的某个值移动到指定的db下

randomkey()

随机获取一个redis的name（不删除）

type(name)

获取name对应值的类型

scan(cursor=0, match=None, count=None)

scan_iter(match=None, count=None)

同字符串操作，用于增量迭代获取key

管道

redis-py默认在执行每次请求都会创建（连接池申请连接）和断开（归还连接池）一次连接操作，如果想要在一次请求中指定多个命令，则可以使用pipline实现一次请求指定多个命令，并且默认情况下一次pipline是原子性操作。

#!/usr/bin/env python3

import redis
pool = redis.ConnectionPool(host='127.0.0.1', port=6379)

r=redis.Redis(connection_pool=pool)

pipe = r.pipeline(transaction=True)

pipe.set('d01','dingyi')
pipe.set('lalala', 'hahaha')

pipe.execute()

发布订阅

发布者：服务器

订阅者：Dashboad和数据处理

Demo如下：

#!/usr/bin/env python3

import redis

class RedisHelper(object):
def __init__(self):
self.__conn = redis.Redis(host='127.0.0.1')
self.chan_sub = 'fm99.9'
self.chan_pub = 'fm99.9'

def public(self, msg):
#向接收端发送消息
self.__conn.publish(self.chan_pub, msg)
return True

def subscribe(self):
#订阅频道
pub = self.__conn.pubsub()
pub.subscribe(self.chan_sub)
#解析来自服务端的想响应
pub.parse_response()
return pub

订阅者

#!/usr/bin/env python3

from RedisHelper import RedisHelper
#实例化
obj = RedisHelper()
#订阅频道，监听服务端
redis_sub = obj.subscribe()

while True:
#接受发布者发来的消息
msg = redis_sub.parse_response()
print(msg)

发布者

#!/usr/bin/env python3

from RedisHelper import RedisHelper
#实例化
obj = RedisHelper()
#发布消息
obj.public('hello')

RabbitMQ

RabbitMQ是一个在AMQP基础上完整的，可复用的企业消息系统，他遵循Mozilla Public License开源协议。

MQ全称为Message Queue，消息队列(MQ)是一种应用程序对应程序的通信方法。应用程序应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来连接他们。消息传递指的是程序之间通过在消息队列中发送数据进行通信而不是直接点用彼此来通信，直接调用通常是用于例如远程过程调用的技术。排队指的是应用程序通过队列来通信队列的使用除了接收和发送应用程序同时执行的要求。

RabbitMQ安装

安装RabbitMQ
apt-get install rabbitmq-server
安装API
pip3 install pika

使用API操作RabbitMQ

基于Queue实现生产者消费者模型

#!/usr/bin/env python3
import queue
import threading

message = queue.Queue(10)

def producer(i):
#   while True:
message.put(i)
print('放入：',i)

def consumer(i):
#   while True:
a = message.get()
print('取出：',a)

for i in range(12):
t = threading.Thread(target=producer, args=(i,))
t.start()

for i in range(10):
t = threading.Thread(target=consumer, args=(i,))

对于RabbitMQ来说，生产和消费不在针对内存里的一个queue对象而是某台服务器上的RabbitMQ-server实现的消息队列

生产者端

#!/usr/bin/env python3
import pika

#生产者
#建立一个阻塞的连接
connection = pika.BlockingConnection(pika.ConnectionParameters(host = '127.0.0.1'))
#生成一个管道，在管道中声明不同的队列
channel = connection.channel()
#在管道中创建一个队列，队列名是hello
channel.queue_declare(queue='hello')
#向队列里发数据，routing_key是队列名，body是内容
#RabbitMQ不能把消息直接发送到队列里，exchange相当于路由器的功能，将内容转发到队列中，exchange起到过滤数据额作用。控制那些数据转发到那些队列中。这里设置为空就是不过滤的意思。
channel.basic_publish(
exchange='',
routing_key='hello',
body='Hello World!'
)
print('[x] sent hello world!')
connection.close()

消费者端

#!/usr/bin/env python3
import pika
#消费者
#建立一个阻塞的连接
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))
#创建一个管道
channel = connection.channel()
#这里也定义了一个队列，虽然在生产者端创建了这个队列，但是如果消费者端先调起，这是的服务器中还没有hello队列，这边就会报
错，再次创建一下这个队列，避免引发错误。
channel.queue_declare(queue='hello')

#ch, method, properties是规定的参数，必须写
def callback(ch, method, properties, body):
print('[x] Received %r' % body)

#消费操作，如果接受到了消息，就调用callback函数，no_ack=True表示不需要确认，如果是False表示会向发送方发送一个确认，确认消息被处理完了。如果消费者遇到情况挂掉了，那么，RabbitMQ会重新将该任务添加到队列中。
channel.basic_consume(callback, queue='hello', no_ack=True)

print('[*] Waiting for messages. To exit press CTRL+C')
#开始阻塞，接受数据
channel.start_consuming()

Work Queues

这种模式下，RabbitMQ默认会把P打的消息依次分发给各个消费者（c），类似负载均衡

生产者

#!/usr/bin/env python3

import pika
import time
import sys
#建立一个阻塞的连接
connection = pika.BlockingConnection(pika.ConnectionParameters('127.0.0.1'))
#创建实例
channel = connection.channel()
#声明一个队列
channel.queue_declare(queue='task_queue')
#定义要传递的信息
message = ''.join(sys.argv[1:]) or "hello world %s" %time.time()
#自动轮询，不需要特殊的语句
channel.basic_publish(
exchange = '',
routing_key='task_queue',#关键字名就是队列名
body = message,
properties=pika.BasicProperties(
#设置数据持久化
delivery_mode=2,
)
)

print("[x] sent %r" % message)
connection.close()

消费者

#!/usr/bin/env python3

import pika
import time
#建立一个阻塞的连接
connection = pika.BlockingConnection(pika.ConnectionParameters('127.0.0.1'))
#创建实例
channel = connection.channel()
#生成一个队列，避免报错
channel.queue_declare(queue='task_queue')

#设置callback方法
def callback(ch, method, properties, body):
print('[x] Receive %r' % body)
time.sleep(20)
print('[x] Done')
print('method.delivery_tag', method.delivery_tag)
#对应生产者端设置消息持久化
ch.basic_ack(delivery_tag=method.delivert_tag)

#消费操作的一些定义
channel.basic_consume(
callback,
queue='task_queue',
no_ack=True,
)

print('[*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

消息不丢失

为了确保信息不会丢失，RabbitMQ支持消息确认功能，让生产者向消费者发送信息的时候，消费者会向生产者返回一个信息，告诉生产者消息已经被处理了。

如果消费者这边有异常（死亡了），在发送消息的时候，RabbitMQ会将没有处理的信息从新放入到队列中。如果这时候有其他的消费者，RabbitMQ就会将消息发送到其他的消费者那里去。这样即使是其中一个消费者有异常（死亡）了，数据也不会消失

在没有设置超时时间的时候，RabbitMQ将在接受者死亡的时候从新传递消息，即使处理信息需要很长很长的时间。

在默认情况下，消息确认功能是被打开的

消息持久化

如果RabbitMQ-server挂掉了，就会在成数据的丢失，避免这种情况，可以再定义队列的时候这样定义

channel。queue_declare(queue='hello', durable=True)

一旦先生成了一个不支持持久化的队列，它是不能再改成支持持久化的。持久化队列必须一开始就要定义好。

在生产者和消费者的代码中，有需要声明一个可以持久化的队列。

上述只是将队列持久化。要实现消息持久化还是需要下面的设置

生产者端
channel.basic_publish(
exchange = '',
routing_key='task_queue',
body = message,
properties=pika.BasicProperties(
#设置数据持久化
delivery_mode=2,
)
)

消费者端
ch.basic_ack(delivery_tag=method.delivert_tag)

消息公平分发

如果Rabbit只管按顺序把消息发到各个消费者上，不考虑消费者负载的话，很可能出现，一个机器配置不高，消息处理不完，另一台配置高的确跟轻松。为了解决此问题，可以设置perfetch=1，意思就是消费者当前的消息还没有处理完，不要再给消费者发送新的消息了。

channel.basic_qos(prefetch_count=1)

带消息持久化+公平分发的完整代码

生产者端

#!/usr/bin/env python3
#生产者
import pika, sys
#创建一个阻塞的连接
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))
#生成管道
channel = connection.channel()
#定义队列，设置为持久化队列,持久化队列必须是新定义的，不能再重新定义之前已经存在的
channel.queue_declare(queue='task_queue_01', durable=True)

message = ''.join(sys.argv[1:]) or 'Hello world!'

#发送消息的操作
channel.basic_publish(
exchange = '',
routing_key='task_queue',
body=message,
#设置消息是持久化的
properties=pika.BasicProperties(
delivery_mode=2,
)
)
print('[x] Sent %r' % message)
connection.close()

消费者端

#!/usr/bin/env python3
import pika, time
#消费者端

#定义一个阻塞的连接
connection = pika.BlockingConnection(pika.ConnectionParameters(host = '127.0.0.1'))
#定义一个管道
channel = connection.channel()
#定义一个持久化队列，避免出错
channel.queue_declare(queue='task_queue_01', durable=True)

print('[*] Waiting for message. To exit press CTRL+C')
#消费的相关动作
def callback(ch, method, properties, body):
print('[x] Received %r' % body)
time.sleep(body.count(b'.'))
print('[x] Dnoe')
#设置消息的持久化
ch.basic_ack(delivery_tag = method.delivery_tag)
#设置当消费者没有处理完信息的时候，不在接受信息
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback, queue='task_queue')

channel.start_consuming()

消息发布与订阅

要实现消息发布的功能，就要用到exchange。

exchange在定义的时候是有类型的，以决定到底是那些Queue符合条件，可以接受消息

fanout：所有bind到此exchange的queue都可以接受消息

direct；通过routingKey和exchange决定的那个唯一的queue可以接受消息

topic：所有符合routingKey（此时可以是个表达式）的routingKey所bind的queue可以接受消息

表达式符号说明：
#代表一个或多个字符
*代表任何字符
例：
#.a会匹配a.a,aa.a, aaa.a等
*.a会匹配a.a,b.a,c.a等
注：使用RoutingKey为#，Exchange Type为topic的时候相当于使用fanout

headers：通过headers来决定吧消息发给哪些queue



1、发向所有和全部接收

消息发布方

#!/usr/bin/env python3
#消息发布方

import pika, sys

connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))

channel = connection.channel()

#因为发布方是向所有队列发送消息，所有不用管声明队列
#exchange='logs'是exchange的名字，type='fanout'是类型
channel.exchange_declare(exchange='logs', type='fanout')

message = ''.join(sys.argv[1:]) or 'hello world'

channel.basic_publish(
exchange='logs',
#如果是fanout的话，routing_key没有用：
routing_key='',
body=message,
)
print('[x] Sent %r ' % message)
connection.close()

消息订阅方

#!/usr/bin/env python3
#消息订阅方

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))

channel = connection.channel()
#订阅方也需要一个同样的声明
channel.exchange_declare(exchange='logs',type='fanout')

#不指定queue名字，rabbit会随机分配一个名字，exclusive=True会在使用此queue的消费者断开后，自动将queue删除
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

#把queue绑定到exchange才能收消息
channel.queue_bind(exchange='logs',queue=queue_name)

print('[*] Waiting for logs. To exit press CTRL+C')

def callback(ch, method, properties, body):
print("[x] %r" % body)

channel.basic_consume(
callback,
queue=queue_name,
no_ack=True
)

channel.start_consuming()

2、有选择的接收消息

RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据很具关键字发送到消息exchange，exchange根据关键字判定应该将数据发送至指定队列。



发布者

#!/usr/bin/env python3
#发布端
import pika, sys
#创建一个阻塞的连接
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))
#声明一个管道
channel = connection.channel()
#设置exchange的类型，名字
channel.exchange_declare(exchange='direct_log', type='direct')
#在脚本的后面第一个参数是针对哪个关键字发送的信息，如果没有则默认是info关键字
severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
#将脚本的第二个参数设定为发送的内容，没有默认为hello world
message = ''.join(sys.argv[2:]) or 'hello world'
#发送信息的动作
channel.basic_publish(
exchange = 'direct_log',
routing_key = severity,#添加关键字
body=message,
)

print("[x] Sent %r:%r" % (severity, message))

connection.close()

订阅者

#!/usr/bin/env python3
#订阅端
import pika, sys
#创建一个阻塞的连接
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))
#声明一个管道
channel = connection.channel()
#声明exchange的名字和类型
channel.exchange_declare(exchange='direct_log', type='direct')
#随机创建一个
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

#客户端这边绑定的关键字,可以是多个
severities = sys.argv[1:]
#不过没有输入关键字的时候弹出错误信息
if not severities:
sys.stderr.write('usage: %s [info] [warring] [error] \n' % sys.argv[0])
sys.exit(1)

#循环关键字列表
for severity in severities:
#绑定队列
channel.queue_bind(
exchange='direct_log',
queue = queue_name,#随机的队列名
routing_key=severity #对应的关键字
)

print('[*] Waiting for logs. To exit press CTRL+C')

#打印收到的内容以及关键字的名字
def callback(ch, method, properties, body):
print('[x] %r:%r' % (method.routing_key, body))
#等待接收数据
channel.basic_consume(callback, queue=queue_name, no_ack=True)

channel.start_consuming()

3、更细致的消息过滤

虽然上面介绍了一些过滤，但是它们仍然后局限性，它们不能标准的进行过滤。在日志系统中，我们不能光注意日志的级别(info、warring、error)，也要注意日志原(mysql的或是Apache的)。



发布端

#!/usr/bin/env python3
import pika, sys
#发布者端
#创建一个阻塞的连接
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))
#声明一个管道
channel = connection.channel()
#声明exchange的名字和种类
channel.exchange_declare(
exchange = 'topic_logs',
type='topic',
)
#设置绑定的关键字，没有的话默认anonyous.info
routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
#获取发送的内容
message = ''.join(sys.argv[2:]) or 'Hello World!'
#发布动作
channel.basic_publish(
exchange='topic_logs',
routing_key=routing_key,
body=message,
)

print("[x] Sent %r:%r" % (routing_key, message))

connection.close()

接收端

#!/usr/bin/env python3
#发布端
import pika, sys
#创建一个阻塞的连接
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))
#声明一个管道
channel = connection.channel()
#声明exchange的名字和类型
channel.exchange_declare(
exchange = 'topic_logs',
type = 'topic'
)
#随机生成队列的名字，并在连接断开打的时候删除该队列
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
#存放绑定的关键字
binding_keys = sys.argv[1:]
#如果关键字没有的时候
if not binding_keys:
sys.stderr.write('Usage: %s [binding_key] ...\n' % sys.argv[0])
sys.exit(1)
#循环获取单个关键字
for binding_key in binding_keys:
#绑定关键字
channel.queue_bind(
exchange = 'topic_logs',
queue = queue_name,
routing_key = binding_key
)

print('[*] Waiting for logs. To exit press CTRL+C')
#定义callback
def callback(ch, method, properties, body):
print("[x] %r:%r" % (method.routing_key, body))
#等到接收信息
channel.basic_consume(
callback,
queue = queue_name,
no_ack=True,
)

channel.start_consuming()

发布方可以通过通配符来选择，来让特定的关键字接收数据，通配符在上面有讲解

Remote procedure call (RPC)

RPC类似一个远程方法调用。客户端发一个请求给服务器端，调用服务器端的一个命令，服务器端返回命令的执行结果。不单要发消息还要返回结果。

服务器端

#!/usr/bin/env python3
#服务器端
import pika, time

connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))

channel = connection.channel()
#声明一个队列
channel.queue_declare(queue='rpc_queue')

#生成一个斐波那契数列
def fib(n):
if n == 0:
return 0
elif n == 1:
return 1
else:
return fib(n-1) + fib(n-2)

#
def on_request(ch, method, props, body):
n = int(body)
print("[.] fib(%s)" % n)
response = fib(n)
#发送数据
ch.basic_publish(
exchange='',
#发送到这个queue中
routing_key=props.reply_to,#这个是客户端发过来的那个随机名字的queue的名字
#发送数据时，附带的标识符，其实队列的名字已经是随机生成唯一的了，这里的标识符唯一性不是为了定位是哪个队列，是为了定位哪条信息，客户端可能给服务端发送多条命令
properties=pika.BasicProperties(
correlation_id=props.correlation_id),
#发送的内容
body=str(response),
)
#持久化
ch.basic_ack(delivery_tag = method.delivery_tag)
#公平分发
channel.basic_qos(prefetch_count=1)
#在接收到数据的时候，执行on_request的操作
channel.basic_consume(on_request, queue='rpc_queue')

print('[x] Awaiting RPC requests')

channel.start_consuming()

客户端

#!/usr/bin/env python3
#客户端
import pika, uuid

class FibonacciRpcClient(object):
def __init__(self):
#创建一个阻塞的连接
self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))
#声明一个管道
self.channel = self.connection.channel()
#生产一个随机名字的队列
result = self.channel.queue_declare(exclusive=True)
self.callback_queue = result.method.queue
#发送信息的操作，self.on_response这里相当于之前例子中的callback
self.channel.basic_consume(
self.on_response, #在下面有定义
no_ack=True,#不阻塞
queue = self.callback_queue,#队列的名字
)
#callback，接收数据的时候执行的操作
def on_response(self, sh, method, props, body):
#props.correlation_id是一个标识符，客户端通过rabbitMQ中的队列，将请求发送给服务端，服务端返回的时候，不能用发送请求的队列，这样会出错，所以客户端在发送请求的时候，同时也会发送一个标识符，告诉服务端用这个队列返回结果。
#corr_id和correlation_id都在下面的call中定义
#初始化的时候，先监听队列等待服务端返回数据，在调用call向服务端发送数据，然后服务端再返回数据，返回的数据如果标识符是一样的，就把返回的数据赋值给self.response
if self.corr_id == props.correlation_id:
self.response = body

def call(self, n):
#用来存储服务端返回的结果
self.response = None
#生成一个随机数，用来做标识符
self.corr_id = str(uuid.uuid4())
#发送信息的操作，在前面的例子中，用于生产者端
self.channel.basic_publish(
exchange='',
#把消息发到了rpc_queue里
routing_key='rpc_queue',
#返回的时候，返回到随机名字的queue(callback_queue)里，标识符是self.corr_id
properties = pika.BasicProperties(
reply_to = self.callback_queue,
correlation_id = self.corr_id,
),
#发送的内容
body = str(n)
)

while self.response is None:
#不断的queue里接收数据，接收到数据会执行on_response操作
self.connection.process_data_events()
#将返回的结果返回
return int(self.response)

#初始化
fibonacci_rpc = FibonacciRpcClient()

print('[x] Requesting fib(30)')
#获取返回的结果
response = fibonacci_rpc.call(30)
print('[.] Got %r' % response