C语言基础学习PYTHON——基础学习D11

20180908内容纲要：

1、RabbitMQ消息队列

(1)RabbitMQ安装

(2)Rabbits示例

模式一：fanout

模式二：direct

模式三：topic

Remote Procedure Call(RPC)

2、Redis缓存

(1)连接方式

(2)连接池

(3)操作

1、string操作

2、hash操作

3、list操作

4、set集合操作

5、其他常用操作

(4)发布订阅

3、小结

4、练习

1 RabbitMQ消息队列

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

MQ的消费-生产者模型的一个典型的代表，一端往消息队列中不断的写入消息，而另一端则可以读取或者订阅队列中的消息。

RabbitMQ的结构图如下：

Broker：简单来说就是消息队列服务器实体。

Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。

Queue：消息队列载体，每个消息都会被投入到一个或多个队列。

Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。

Routing Key：路由关键字，

exchange根据这个关键字进行消息投递。

vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。

producer：消息生产者，就是投递消息的程序。

consumer：消息消费者，就是接受消息的程序。

channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。

当然，也有其他消息队列，比如ZeroMQ、ActiveMQ等。

(1)RabbitMQ的安装

首先，因为RabbitMQ由ERLANG实现，下载ERLANG 源代码。

然后，下载RabbitMQ。

在windows安装中可能会出现这样的问题：RabbitMQ报错Error: unable to connect to node rabbit@xxx: nodedown

(2)RabbitMQ基本示例

a、消息分发轮询

1 #Author：ZhangKanghui

2

3 importpika4

5 #建立一个简单的socket

6 connection =pika.BlockingConnection(7 pika.ConnectionParameters('localhost')8 )9 #声明一个管道

10 channel =connection.channel()11

12 #声明queue

13 channel.queue_declare(queue="hello")14

15 channel.basic_publish(exchange="",16 routing_key='hello', #queue名字

17 body='Hello world',)18

19 print("[x] send 'hello world'")20 connection.close()

RabbitMQ_send

1 #Author：ZhangKanghui

2

3 importpika4 importtime5

6 #建立一个简单的socket

7 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))8 #声明一个管道

9 channel =connection.channel()10

11 #you may ask why declare the queue again?

12 #we could avoid that if we were sure that the queue already exits.

13 #如果接受端先启动不会报错，如果没有声明，先启动发送端在启动接受端不会出错。但是如果先启动接受端就会有问题

14 channel.queue_declare(queue="hello")15

16 defcallback(ch,method,properties, body):17 #这个callback函数中的四个参数分别是什么呢？我们打印一下

18 print("--->>>",ch,method,properties,body)19 time.sleep(30)20 print("[x] Received %r"%body)21 ch.basic_ack(delivery_tag=method.delivery_tag)22

23 channel.basic_qos(prefetch_count=1)24 channel.basic_consume(callback, #消费消息。如果收到消息就调用callback，

25 queue='hello',26 no_ack=True) #No acknowledgement

27 #如果一个发送端，多个接受端呢？

28 #如果no_ack存在，当消息正在处理的过程中接受端断开，消息就会丢失。发送端发送的数据就会删除。

29 #如果no_ack不存在，发送端就会等待接受端处理完毕的指令。如果此时接受端1断开，那么就由2接收，如果都断开，

30 #那么数据会一直存储在发送端，直至有新的接受端出现，并且发送端收到接受端数据处理完毕的指令，数据才会在发送端被删除

31

32 print('[*] waiting for message .To exit')33 channel.start_consuming()

RabbitMQ_receive

在实现一个消费者对应一个生产者，那能不能一对多呢？

在一个生产者对应多个消费者的时候，生产者发出的消息会根据启动顺序依次被消费者接收。即消息分发轮询。

如果no_ack存在，当消息正在处理的过程中接受端断开，消息就会丢失。发送端发送的数据就会删除；

如果no_ack不存在，发送端就会等待接受端处理完毕的指令。如果此时接受端1断开，那么就由2接收，如果都断开；

那么数据会一直存储在发送端，直至有新的接受端出现，并且发送端收到接受端数据处理完毕的指令，数据才会在发送端被删除。

Message acknowledgments are turned on by default. In previous examples we explicitly turned them off via the no_ack=True flag.

消息确认在默认情况下是打开的。在前面的示例中，我们通过no_ack=True标志显式地关闭了它们。

It's time to remove this flag and send a proper acknowledgment from the worker, once we're done with a task.

当我们完成一项任务时，是时候移除此标志并从工作人员发送适当的确认信息了。

如果客户端突然断开，数据则会丢失，为了防止数据丢失需要下面一段代码：

1 defcallback(ch,method,properties, body):2 print("--->>>",ch,method,properties,body)3 time.sleep(30)4 print("[x] Received %r"%body)5 ch.basic_ack(delivery_tag=method.delivery_tag)

ch.basic_ack(delivery_tag=method.delivery_tag)

b、消息持久化

在RabbitMQ的安装路径下C:\Program Files\RabbitMQ Server\rabbitmq_server-3.6.5\sbin>在sbin文件夹中rabbit的一些指令

比如rabbitmqctl.bat list_queues可以查询存在的队列消息等。

那么客户端的断开会发生数据丢失，通过确认函数来保证数据不会丢失。那么如果服务器端断开呢？数据一样会丢失。

为了防止服务器端断开导致数据丢失，我们需要以下两步：

durable队列持久化　　　　　　delievery消息持久化

1、首先要进行声明。

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

2、虽然在这里声明，但并不能真的持久，还需这样。

channel.queue_declare(queue='task_queue', durable=True)

这样，即使RabbitMQ重新启动，task_queue队列也不会丢失。

1 channel.basic_publish(exchange='',2 routing_key="task_queue",3 body=message,4 properties=pika.BasicProperties(5 delivery_mode = 2, #make message persistent

6 ))

消息持久化

c、消息公平分发

如果Rabbit只管按顺序把消息发到各个消费者身上，不考虑消费者负载的话，很可能出现，一个机器配置不高的消费者那里堆积了很多消息处理不完，同时配置高的消费者却一直很轻松。那么解决办法？

channel.basic_qos(prefetch_count=1)

这行代码只需加入到消费者端即可。

1 #Author：ZhangKanghui

2

3 importpika4 importtime5

6 #建立一个简单的socket

7 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))8 #声明一个管道

9 channel =connection.channel()10

11 #you may ask why declare the queue again?

12 #we could avoid that if we were sure that the queue already exits.

13 #如果接受端先启动不会报错，如果没有声明，先启动发送端在启动接受端不会出错。但是如果先启动接受端就会有问题

14 channel.queue_declare(queue="hello")15

16 defcallback(ch,method,properties, body):17 #这个callback函数中的四个参数分别是什么呢？我们打印一下

18 print("--->>>",ch,method,properties,body)19 time.sleep(30)20 print("[x] Received %r"%body)21 ch.basic_ack(delivery_tag=method.delivery_tag)22

23 channel.basic_qos(prefetch_count=1) #消息分发

24 channel.basic_consume(callback, #消费消息。如果收到消息就调用callback，

25 queue='hello',26 no_ack=True) #No acknowledgement

27 #如果一个发送端，多个接受端呢？

28 #如果no_ack存在，当消息正在处理的过程中接受端断开，消息就会丢失。发送端发送的数据就会删除。

29 #如果no_ack不存在，发送端就会等待接受端处理完毕的指令。如果此时接受端1断开，那么就由2接收，如果都断开，

30 #那么数据会一直存储在发送端，直至有新的接受端出现，并且发送端收到接受端数据处理完毕的指令，数据才会在发送端被删除

31

32 print('[*] waiting for message .To exit')33 channel.start_consuming()

持久化+消息分发

d、消息发布/订阅

如果服务器发出一条消息，是否所有的客户端都能收到消息呢？

Exchange在定义的时候是有类型的，以决定到底是哪些Queue符合条件，可以接收消息。

fanout:所有bind到此exchange的queue都可以接收消息；direct:通过routingKey和exchange决定的那个唯一的queue可以接收消息；topic:所有符合routingKey(此时可以是一个表达式)的routingKey所bind的queue可以接收消息；

headers: 通过headers 来决定把消息发给哪些queue　　　　这个不常用。

表达式符号说明：#代表一个或多个字符，*代表任何字符      例：#.a会匹配a.a，aa.a，aaa.a等          *.a会匹配a.a，b.a，c.a等

模式一：fanout广播

1 #Author：ZhangKanghui

2

3 importpika4 importsys5

6 connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))7 channel =connection.channel()8

9 channel.exchange_declare(exchange='logs',10 type='fanout')11 message = ''.join(sys.argv[1:]) or "info:hello world！"

12 channel.basic_publish(exchange='logs',13 routing_key='',14 body =message)15 print("[x] send %r"%message)16 connection.close()

fanout_publisher

1 #Author：ZhangKanghui

2

3 importpika4

5 connection =pika.BlockingConnection(pika.ConnectionParameters(6 host='localhost'))7 channel =connection.channel()8

9 channel.exchange_declare(exchange='logs',10 type='fanout')11 result = channel.queue_declare(exclusive=True) #exclusive排他的，唯一的

12 #在这里声明队列的原因？

13 #不指定queue的名字，rabbit会随机分配一个名字，exclusive=True会使次queue的消费者断开后，自动将queue删除

14 queue_name =result.method.queue15 print(queue_name)16

17 channel.queue_bind(exchange='logs',18 queue=queue_name)19 print('[*] waiting for logs.to exit')20

21 defcallback(ch,method,properties,body):22 print("[x] %r"%body)23

24

25 channel.basic_consume(callback,26 queue=queue_name,27 no_ack=True)28 channel.start_consuming()

fanout_subscriber

模式二：direct有选择的接收消息

1 #Author：ZhangKanghui

2

3 importpika4 importsys5

6 connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))7 channel =connection.channel()8

9 channel.exchange_declare(exchange='logs',10 type='direct')11

12 severity = sys.argv[1:] if len(sys.argv) > 1 else 'info'

13 message = ''.join(sys.argv[2:]) or "info:hello world！"

14

15 channel.basic_publish(exchange='direct_logs',16 routing_key=severity,17 body =message)18 print("[x] send %r:%r"%(severity,message))19 connection.close()

direct_publisher

1 #Author：ZhangKanghui

2

3 importpika4 importsys5

6 connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))7 channel =connection.channel()8

9 channel.exchange_declare(exchange='direct_logs',10 type='direct')11 result = channel.queue_declare(exclusive=True)12 queue_name =result.method.queue13

14 severities = sys.argv[1:]15 if notseverities:16 sys.stderr.write("usage:%s [info] [warining] [error]\n"%sys.argv[0])17 sys.exit(1)18 print(severities)19 for severity inseverities:20 channel.queue_bind(exchange='direct_logs',21 queue=queue_name,22 routing_key=severity)23 channel.basic_consume(callback,24 queue=queue_name,25 no_ack=True)26 channel.start_consuming()

direct_subscriber

模式三：topic更细致的消息过滤

*.orange.*所有中间包含orange

*.*.rabbit所有以rabbit结尾的

lazy.#所有以lazy开头的

1 #Author：ZhangKanghui

2

3 importpika4 importsys5

6 connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))7 channel =connection.channel()8

9 channel.exchange_declare(exchange='topic_logs',10 type='topic')11

12 routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous info'

13 message = ''.join(sys.argv[2:]) or "info:hello world！"

14 channel.basic_publish(exchange='direct_logs',15 routing_key=routing_key,16 body =message)17 print("[x] send %r:%r"%(routing_key,message))18 connection.close()

topic_publisher

1 #Author：ZhangKanghui

2

3 importpika4 importsys5

6 connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))7 channel =connection.channel()8

9 channel.exchange_declare(exchange='topic_logs',10 type='topic')11

12 result = channel.queue_declare(exclusive=True)13 queue_name =result.method.queue14

15 binding_keys = sys.argv[1:]16 if notbinding_keys:17 sys.stderr.write("Usage:%s [binding_key] ...\n"%sys.argv[0])18 sys.exit(1)19

20 print("[*] waiting for logs .to exit")21

22 defcallback(ch,method,propertities,body):23 print("[x] %r:%r"%(method.routing_key,body))24

25 channel.basic_consume(callback,26 queue=queue_name,27 no_ack=True)28 channel.start_consuming()

topic_subscriber

To receive all the logs run:

python receive_logs_topic.py "#"

To receive all logs from the facility "kern":

python receive_logs_topic.py "kern.*"

Or if you want to hear only about "critical" logs:

python receive_logs_topic.py "*.critical"

You can create multiple bindings:

python receive_logs_topic.py "kern.*" "*.critical"

And to emit a log with a routing key "kern.critical" type:

python emit_log_topic.py "kern.critical" "A critical kernel error"

Remote procedure call (RPC)远程过程调用

RPC(Remote Procedure Call Protocol)——远程过程调用协议，它是一种通过网络从远程计算机程序上请求服务，而不需要了解底层网络技术的协议。该协议允许运行于一台计算机的程序调用另一台计算机的子程序，而程序员无需额外地为这个交互作用编程。RPC协议假定某些传输协议的存在，如TCP或UDP，为通信程序之间携带信息数据。在OSI网络通信模型中，RPC跨越了传输层和应用层。RPC使得开发包括网络分布式多程序在内的应用程序更加容易。

.

1 #Author：ZhangKanghui

2

3 importpika4 importtime5

6 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))7 channel =connection.channel()8 channel.queue_declare(queue='rpc_queue')9

10 deffib(n):11 if n ==0:12 return013 elif n == 1:14 return 1

15 else:16 return fib(n-1) + fib(n-2)17

18 defon_request(ch,method,props,body):19 n =int(body)20 print("[.] fib (%s)"%n)21 reponse =fib(n)22 ch.basic_publish(exchange='',23 routing_key=props.rely_to,24 properties=pika.BasicProperties(correlation_id=props.correlation_id),25 body=str(reponse))26 ch.basic_ack(delivery_tag=method.delivery_tag)27

28 channel.basic_qos(prefetch_count=1)29 channel.basic_consume(on_request,queue='rpc_queue')30

31 print("[x] waiting rpc request")32 channel.start_consuming()

rpc_server

1 #Author：ZhangKanghui

2

3 importpika4 importuuid5 importtime6

7 classFibonacciRpcClient(object):8 def __init__(self):9 self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))10 self.channel =self.connection.channel()11 result = self.channel.queue_declare(exclusive=True)12 self.callback_queue =result.method.queue13 self.channel.basic_consume(self.on_response,no_ack=True,queue=self.callback_queue)14

15 defon_response(self,ch,method,props,body):16 if self.corr_id ==props.correlation_id:17 self.response =body18

19 defcall(self,n):20 self.response =None21 self.corr_id =str(uuid.uuid4())22 self.channel.basic_publish(exchange='',23 routing_key='rpc_queue',24 properties=pika.BasicProperties(reply_to=self.callback_queue,25 correlation_id=self.corr_id),26 body=str(n))27 while self.response isNone:28 self.connection.process_data_events() #相当于非阻塞的start_consuming

29 print("no msg...")30 time.sleep(0.5)31 returnint(self.response)32

33 fibonacci_rpc =FibonacciRpcClient()34

35 print("[x] Requesting fib (30)")36 reponse = fibonacci_rpc.call(30)37 print("[.] Got %r"% reponse)

rpc_client

2 Redis缓存

Redis是一个Key-Value存储系统。和Memcached类似，它支持存储的value类型相对更多。

包括string(字符串)、list(链表)、set(集合)、zset(sorted set --有序集合)和hash(哈希类型)。

这些数据类型都支持push/pop、add/remove及取交集并集和差集及更丰富的操作，而且这些操作都是原子性的。在此基础上，redis支持各种不同方式的排序。

与memcached一样，为了保证效率，数据都是缓存在内存中。

区别的是redis会周期性的把更新的数据写入磁盘或者把修改操作写入追加的记录文件，并且在此基础上实现了master-slave(主从)同步。

memcached 的使用：

Windows下安装Redis教程：

(1)连接方式

redis-py提供两个类Redis和StrictRedis用于实现Redis的命令，StrictRedis用于实现大部分官方的命令，并使用官方的语法和命令.

Redis是StrictRedis的子类，用于向后兼容旧版本的redis-py。

#Author：ZhangKanghui

importredis

r= redis.Redis(host='localhost',port=6379)

r.set('foo','bar')

r.get('foo')print(r.get('foo'))

redis_connection

(2) 连接池

redis-py使用connection pool来管理对一个redis server的所有连接，避免每次建立、释放连接的开销。默认，每个Redis实例都会维护一个自己的连接池。

可以直接建立一个连接池，然后作为参数Redis，这样就可以实现多个Redis实例共享一个连接池。

1 #Author：ZhangKanghui

2

3 importredis4

5 pool = redis.ConnectionPool(host='localhost',port=6379)6

7 r = redis.Redis(connection_pool=pool)8 r.set('foo','bar')9 r.get('foo')10 print(r.get('foo'))

redis_connection pool

(3) 操作

1、string操作

redis中的String在在内存中按照一个name对应一个value来存储。如图：

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

1 在Redis中设置值，默认，不存在则创建，存在则修改2 参数：3 ex，过期时间(秒)4 px，过期时间(毫秒)5 nx，如果设置为True，则只有name不存在时，当前set操作才执行6 xx，如果设置为True，则只有name存在时，岗前set操作才执行

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

setnx(name, value)

1 设置值，只有name不存在时，执行设置操作(添加)

setnx(name, value)

setex(name, value, time)

1 #设置值

2 #参数：

3 #time，过期时间(数字秒 或 timedelta对象)

setex(name, value, time)

psetex(name, time_ms, value)

1 #设置值

2 #参数：

3 #time_ms，过期时间(数字毫秒 或 timedelta对象)

psetex(name, time_ms, value)

mset(*args, **kwargs)

1 批量设置值2 如：3 mset(k1='v1', k2='v2')4 或5 mget({'k1': 'v1', 'k2': 'v2'})

mset(*args, **kwargs)

get(name)

1 获取值

get(name)

mget(keys, *args)

1 批量获取2 如：3 mget('ylr', 'wupeiqi')4 或5 r.mget(['ylr', 'wupeiqi'])

mget(keys, *args)

getset(name, value)

1 设置新值并获取原来的值

getset(name, value)

getrange(key, start, end)

getrange(key, start, end)

setrange(name, offset, value)

1 #修改字符串内容，从指定字符串索引开始向后替换(新值太长时，则向后添加)

2 #参数：

3 #offset，字符串的索引，字节(一个汉字三个字节)

4 #value，要设置的值

setrange(name, offset, value)

setbit(name, offset, value)

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

2

3 #参数：

4 #name，redis的name

5 #offset，位的索引(将值变换成二进制后再进行索引)

6 #value，值只能是 1 或 0

7

8 #注：如果在Redis中有一个对应： n1 = "foo"，

9 那么字符串foo的二进制表示为：01100110 01101111 01101111

10 所以，如果执行 setbit('n1', 7, 1)，则就会将第7位设置为1，11 那么最终二进制则变成 01100111 01101111 01101111，即："goo"

12

13 #扩展，转换二进制表示：

14

15 #source = "撒拉黑"

16 source = "foo"

17

18 for i insource:19 num =ord(i)20 print bin(num).replace('b','')21

22 特别的，如果source是汉字 "撒拉黑"怎么办？23 答：对于utf-8，每一个汉字占 3 个字节，那么 "撒拉黑"则有 9个字节24 对于汉字，for循环时候会按照 字节 迭代，那么在迭代时，将每一个字节转换 十进制数，然后再将十进制数转换成二进制25 11100110 10101101 10100110 11100110 10110010 10011011 11101001 10111101 10010000

26 -------------------------- ----------------------------- -----------------------------

27 撒 拉 黑

setbit(name, offset, value)

getbit(name, offset)

1 #获取name对应的值的二进制表示中的某位的值 (0或1)

getbit(name, offset)

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

1 #获取name对应的值的二进制表示中 1 的个数

2 #参数：

3 #key，Redis的name

4 #start，位起始位置

5 #end，位结束位置

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

strlen(name)

1 #返回name对应值的字节长度(一个汉字3个字节)

strlen(name)

incr(self, name, amount=1)

1 #自增 name对应的值，当name不存在时，则创建name＝amount，否则，则自增。

2

3 #参数：

4 #name,Redis的name

5 #amount,自增数(必须是整数)

6

7 #注：同incrby

incr(self, name, amount=1)

incrbyfloat(self, name, amount=1.0)

1 #自增 name对应的值，当name不存在时，则创建name＝amount，否则，则自增。

2

3 #参数：

4 #name,Redis的name

5 #amount,自增数(浮点型)

incrbyfloat(self, name, amount=1.0)

decr(self, name, amount=1)

1 #自减 name对应的值，当name不存在时，则创建name＝amount，否则，则自减。

2

3 #参数：

4 #name,Redis的name

5 #amount,自减数(整数)

decr(self, name, amount=1)

append(key, value)

1 #在redis name对应的值后面追加内容

2

3 #参数：

4 key, redis的name5 value, 要追加的字符串

append(key, value)

2、hash操作

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

hset(name, key, value)

1 #name对应的hash中设置一个键值对(不存在，则创建；否则，修改)

2

3 #参数：

4 #name，redis的name

5 #key，name对应的hash中的key

6 #value，name对应的hash中的value

7

8 #注：

9 #hsetnx(name, key, value),当name对应的hash中不存在当前key时则创建(相当于添加)

hset(name, key, value)

hmset(name, mapping)

1 #在name对应的hash中批量设置键值对

2

3 #参数：

4 #name，redis的name

5 #mapping，字典，如：{'k1':'v1', 'k2': 'v2'}

6

7 #如：

8 #r.hmset('xx', {'k1':'v1', 'k2': 'v2'})

hmset(name, mapping)

hget(name,key)

1 #在name对应的hash中获取根据key获取value

hget(name,key)

hmget(name, keys, *args)

1 #在name对应的hash中获取多个key的值

2

3 #参数：

4 #name，reids对应的name

5 #keys，要获取key集合，如：['k1', 'k2', 'k3']

6 #*args，要获取的key，如：k1,k2,k3

7

8 #如：

9 #r.mget('xx', ['k1', 'k2'])

10 #或

11 #print r.hmget('xx', 'k1', 'k2')

hmget(name, keys, *args)

hgetall(name)

1 获取name对应hash的所有键值

hgetall(name)

hlen(name)

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

hlen(name)

hkeys(name)

1 #获取name对应的hash中所有的key的值

hkeys(name)

hvals(name)

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

hvals(name)

hexists(name, key)

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

hexists(name, key)

hdel(name,*keys)

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

hdel(name,*keys)

hincrby(name, key, amount=1)

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

2 #参数：

3 #name，redis中的name

4 #key， hash对应的key

5 #amount，自增数(整数)

hincrby(name, key, amount=1)

hincrbyfloat(name, key, amount=1.0)

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

2

3 #参数：

4 #name，redis中的name

5 #key， hash对应的key

6 #amount，自增数(浮点数)

7

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

hincrbyfloat(name, key, amount=1.0)

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

Start a full hash scan with:

HSCAN myhash 0

Start a hash scan with fields matching a pattern with:

HSCAN myhash 0 MATCH order_*

Start a hash scan with fields matching a pattern and forcing the scan command to do more scanning with:

HSCAN myhash 0 MATCH order_* COUNT 1000

1 #增量式迭代获取，对于数据大的数据非常有用，hscan可以实现分片的获取数据，并非一次性将数据全部获取完，从而放置内存被撑爆

2

3 #参数：

4 #name，redis的name

5 #cursor，游标(基于游标分批取获取数据)

6 #match，匹配指定key，默认None 表示所有的key

7 #count，每次分片最少获取个数，默认None表示采用Redis的默认分片个数

8

9 #如：

10 #第一次：cursor1, data1 = r.hscan('xx', cursor=0, match=None, count=None)

11 #第二次：cursor2, data1 = r.hscan('xx', cursor=cursor1, match=None, count=None)

12 #...

13 #直到返回值cursor的值为0时，表示数据已经通过分片获取完毕

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

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

1 #利用yield封装hscan创建生成器，实现分批去redis中获取数据

2

3 #参数：

4 #match，匹配指定key，默认None 表示所有的key

5 #count，每次分片最少获取个数，默认None表示采用Redis的默认分片个数

6

7 #如：

8 #for item in r.hscan_iter('xx'):

9 #print item

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

3、list操作

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

lpush(name,values)

1 #在name对应的list中添加元素，每个新的元素都添加到列表的最左边

2

3 #如：

4 #r.lpush('oo', 11,22,33)

5 #保存顺序为: 33,22,11

6

7 #扩展：

8 #rpush(name, values) 表示从右向左操作

lpush(name,values)

lpushx(name,value)

1 #在name对应的list中添加元素，只有name已经存在时，值添加到列表的最左边

2

3 #更多：

4 #rpushx(name, value) 表示从右向左操作

lpushx(name,value)

llen(name)

1 #name对应的list元素的个数

llen(name)

linsert(name, where, refvalue, value))

1 #在name对应的列表的某一个值前或后插入一个新值

2

3 #参数：

4 #name，redis的name

5 #where，BEFORE或AFTER

6 #refvalue，标杆值，即：在它前后插入数据

7 #value，要插入的数据

linsert(name, where, refvalue, value))

r.lset(name, index, value)

1 #对name对应的list中的某一个索引位置重新赋值

2

3 #参数：

4 #name，redis的name

5 #index，list的索引位置

6 #value，要设置的值

r.lset(name, index, value)

r.lrem(name, value, num)

1 #在name对应的list中删除指定的值

2

3 #参数：

4 #name，redis的name

5 #value，要删除的值

6 #num， num=0，删除列表中所有的指定值；

7 #num=2,从前到后，删除2个；

8 #num=-2,从后向前，删除2个

r.lrem(name, value, num)

lpop(name)

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

2

3 #更多：

4 #rpop(name) 表示从右向左操作

lpop(name)

lindex(name, index)

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

lindex(name, index)

lrange(name, start, end)

1 #在name对应的列表分片获取数据

2 #参数：

3 #name，redis的name

4 #start，索引的起始位置

5 #end，索引结束位置

lrange(name, start, end)

ltrim(name, start, end)

1 #在name对应的列表中移除没有在start-end索引之间的值

2 #参数：

3 #name，redis的name

4 #start，索引的起始位置

5 #end，索引结束位置

ltrim(name, start, end)

rpoplpush(src, dst)

1 #从一个列表取出最右边的元素，同时将其添加至另一个列表的最左边

2 #参数：

3 #src，要取数据的列表的name

4 #dst，要添加数据的列表的name

rpoplpush(src, dst)

blpop(keys, timeout)

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

2

3 #参数：

4 #keys，redis的name的集合

5 #timeout，超时时间，当元素所有列表的元素获取完之后，阻塞等待列表内有数据的时间(秒), 0 表示永远阻塞

6

7 #更多：

8 #r.brpop(keys, timeout)，从右向左获取数据

blpop(keys, timeout)

brpoplpush(src, dst, timeout=0)

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

2

3 #参数：

4 #src，取出并要移除元素的列表对应的name

5 #dst，要插入元素的列表对应的name

6 #timeout，当src对应的列表中没有数据时，阻塞等待其有数据的超时时间(秒)，0 表示永远阻塞

brpoplpush(src, dst, timeout=0)

3、set集合操作

Set操作，Set集合就是不允许重复的列表

sadd(name,values)

1 #name对应的集合中添加元素

sadd(name,values)

scard(name)

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

scard(name)

sdiff(keys, *args)

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

sdiff(keys, *args)

sdiffstore(dest, keys, *args)

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

sdiffstore(dest, keys, *args)

sinter(keys, *args)

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

sinter(keys, *args)

sinterstore(dest, keys, *args)

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

sinterstore(dest, keys, *args)

sismember(name, value)

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

sismember(name, value)

smembers(name)

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

smembers(name)

smove(src, dst, value)

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

smove(src, dst, value)

spop(name)

1 #从集合的右侧(尾部)移除一个成员，并将其返回

spop(name)

srandmember(name, numbers)

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

srandmember(name, numbers)

srem(name, values)

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

srem(name, values)

sunion(keys, *args)

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

sunion(keys, *args)

sunionstore(dest,keys, *args)

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

sunionstore(dest,keys, *args)

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

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

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

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

有序集合，在集合的基础上，为每元素排序；元素的排序需要根据另外一个值来进行比较

所以，对于有序集合，每一个元素有两个值，即：值和分数，分数专门用来做排序。

zadd(name, *args, **kwargs)

1 #在name对应的有序集合中添加元素

2 #如：

3 #zadd('zz', 'n1', 1, 'n2', 2)

4 #或

5 #zadd('zz', n1=11, n2=22)

zadd(name, *args, **kwargs)

zcard(name)

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

zcard(name)

zcount(name, min, max)

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

zcount(name, min, max)

zincrby(name, value, amount)

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

zincrby(name, value, amount)

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

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

2

3 #参数：

4 #name，redis的name

5 #start，有序集合索引起始位置(非分数)

6 #end，有序集合索引结束位置(非分数)

7 #desc，排序规则，默认按照分数从小到大排序

8 #withscores，是否获取元素的分数，默认只获取元素的值

9 #score_cast_func，对分数进行数据转换的函数

10

11 #更多：

12 #从大到小排序

13 #zrevrange(name, start, end, withscores=False, score_cast_func=float)

14

15 #按照分数范围获取name对应的有序集合的元素

16 #zrangebyscore(name, min, max, start=None, num=None, withscores=False, score_cast_func=float)

17 #从大到小排序

18 #zrevrangebyscore(name, max, min, start=None, num=None, withscores=False, score_cast_func=float)

r.zrange

zrank(name, value)

1 #获取某个值在 name对应的有序集合中的排行(从 0 开始)

2

3 #更多：

4 #zrevrank(name, value)，从大到小排序

zrank(name, value)

zrem(name, values)

1 #删除name对应的有序集合中值是values的成员

2

3 #如：zrem('zz', ['s1', 's2'])

zrem(name, values)

zremrangebyrank(name, min, max)

1 #根据排行范围删除

zremrangebyrank(name, min, max)

zremrangebyscore(name, min, max)

1 #根据分数范围删除

zremrangebyscore(name, min, max)

zscore(name, value)

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

zscore(name, value)

zinterstore(dest, keys, aggregate=None)

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

2 #aggregate的值为: SUM MIN MAX

zinterstore(dest, keys, aggregate=None)

zunionstore(dest, keys, aggregate=None)

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

2 #aggregate的值为: SUM MIN MAX

zunionstore(dest, keys, aggregate=None)

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

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

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

zscan

4、常用其他操作

delete(*names)

1 #根据删除redis中的任意数据类型

delete(*names)

exists(name)

1 #检测redis的name是否存在

exists(name)

keys(pattern='*')

1 #根据模型获取redis的name

2

3 #更多：

4 #KEYS * 匹配数据库中所有 key 。

5 #KEYS h?llo 匹配 hello ， hallo 和 hxllo 等。

6 #KEYS h*llo 匹配 hllo 和 heeeeello 等。

7 #KEYS h[ae]llo 匹配 hello 和 hallo ，但不匹配 hillo

keys(pattern='*')

expire(name ,time)

1 #为某个redis的某个name设置超时时间

expire(name ,time)

rename(src, dst)

1 #对redis的name重命名为

rename(src, dst)

move(name, db))

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

move(name, db))

randomkey()

1 #随机获取一个redis的name(不删除)

randomkey()

type(name)

1 #获取name对应值的类型

type(name)

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

scan_iter(match=None, count=None)

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

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

(4)发布订阅

1 from monitor.RedisHelper importRedisHelper2

3 obj =RedisHelper()4 redis_sub =obj.subscribe()5

6 whileTrue:7 msg=redis_sub.parse_response()8 print msg

发布者

1 from monitor.RedisHelper importRedisHelper2

3 obj =RedisHelper()4 obj.public('hello')

订阅者

3 小结

路漫漫其修远兮，吾将上下而求索~!

真的是这样，有好多事情，你不经历过，真的是不知道有多niux！

间歇性踌躇满志，持续性混吃等死。

4 练习

题目：rpc命令端

需求：

可以异步的执行多个命令

对多台机器

作业是真的不会做。有机会有勇气可以点进来看看：

我是尾巴~

这次推荐：如何用几个简单的命令改善你的Linux安全

虽不才，才要坚持。