RabbitMQ消息队列
RabbitMQ is the most widely deployed open source message broker.
RabbitMQ消息队列与threading Queue及Process Queue
1.threading Queue: 仅可实现在同一进程内的线程之间的交互；
2.进程Queue：父进程与子进程进行交互，或者同属于同一父进程下多个子进程间的交互；
3.那如何实现两个相互独立的Python进程通信呢？如果是java进程与Python进程的交互呢？->使用RabbitMQ
RabbitMQ的安装-win-详细戳这里
安装erlang->加ERLANG_HOME的用户变量->PATH中添加->%ERLANG_HOME%in->cmd下运行erl出现版本即成功
安装RabbitMQ->进Rabbitmq的cmd->rabbitmq-plugins.bat enable rabbitmq_management->启动rabbitmq-server.bat->登录http://localhost:15672/
RabbitMQ在Python上的clients-详细戳这里
实现最简单的队列通信
applications Produce Messages:
Exchanges Route and Filter Messages:On RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
Queues store and forward Messages:
Applications Consume Messages:
1.在一对多的消息分发时，采用轮询模式，即按照先后顺序，逐个处理produce发的消息；
2.永久保存队列使用channel.queue_declare(queue='hello', durable=True)；
3.永久保存消息使用channel.basic_publish(properties=pika.BasicProperties(delivery_mode=2 ))；
4.client对消息的确认channel.basic_consume(auto_ack=False)，False是默认需要手动确认，True则为自动确认，如果消息得不到确认则会一直停留在MQ中；
5.ch.basic_ack(delivery_tag=method.delivery_tag) # 对消息的确认；
6.channel.basic_qos(prefetch_count=1) 公平分发->Client上配置，如果我有prefetch_count条消息未处理完，你就先不要给我发消息；
最简单的代码-Produce
import pika
connection = pika.BlockingConnection(
pika.ConnectionParameters('localhost') # 相当于建立一个最基本的socket，可以设置很多参数
)
channel = connection.channel() # 相当于声明一个管道
# 声明queue，一个名字叫"hello"的queue
# durable=True 让队列可以永久保存，即server端异常重启后仍然在
channel.queue_declare(queue='hello', durable=True)
channel.basic_publish(exchange='',
routing_key='hello', # queue的名字
body='Hello World!', # 消息内容
properties=pika.BasicProperties(delivery_mode=2 ) # 使消息持久化
)
print(" [x] Sent 'Hello World!'")
connection.close()
最简单的代码-Comsume
import pika
import time
connection = pika.BlockingConnection(pika.ConnectionParameters(
'localhost'))
channel = connection.channel()
# You may ask why we declare the queue again ‒ we have already declared it in our previous code.
# We could avoid that if we were sure that the queue already exists. For example if send.py program
# was run before. But we're not yet sure which program to run first. In such cases it's a good
# practice to repeat declaring the queue in both programs.
channel.queue_declare(queue='hello', durable=True)
def callback(ch, method, properties, body): # ch 声明的管道的内存对象地址 #
#print("-->", ch, method, properties)
print(" [x] Received %r" % body)
time.sleep(30)
print("30s is over!")
ch.basic_ack(delivery_tag=method.delivery_tag) # 对消息的确认
# 公平分发之如果我有条消息未处理完，你就先不要给我发消息
# 用于一对多时，各consume处理能力不同；
channel.basic_qos(prefetch_count=1)
# 消费消息
# 如果收到消息就调用callback函数，来处理消息
channel.basic_consume(queue="hello",
on_message_callback=callback,
auto_ack=False)
# auto_ack是用来决定消息是否自动确认，如果不自动确认则需要手动确认；
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming() # 开始收消息，一直收，不止这一条，会一直卡住
Message的广播
Exchanges Route and Filter Messages：An exchange is a very simple thing. On one side it receives messages from producers and the other side it pushes them to queues. The exchange must know exactly what to do with a message it receives. Should it be appended to a particular queue? Should it be appended to many queues? Or should it get discarded. The rules for that are defined by the exchange type.
Exchange在定义的时候是有类型的，以决定哪些Queue符合条件，可以接收消息：
1.fanout：所有bind到此exchange的queue都可以接收消息；->纯广播模式
2.direct：通过routingKey和exchange决定的哪个唯一的Queue可以接收消息；->根据关键字分流的广播模式
3.topic：所有符合routingKey表达式的，routingKey所bind的queue可以接收消息；->极细致的广播模式
4.headers：通过headers来决定把消息发给哪些queue；
Exchange的fanout类型
1.exchange的fanout的特点是，producer发消息不管有没有queue收，都不会在exchange中等待用户接收；
Producer端：
import pika
# 建立连接，及新建channel
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
# 基于channel创建一个fanout类型的exchange
channel.exchange_declare(exchange='logs', exchange_type='fanout')
message = "Info: Hello World!"
# 因为exchange的type是fanout即广播模式，则不需要特定绑定队列，即routing_key=''
channel.basic_publish(exchange='logs', routing_key='', body=message)
print(" [x] Sent %r" % message)
connection.close()
Consumer端：
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
# 基于channel创建一个fanout类型的exchange
channel.exchange_declare(exchange='logs', exchange_type='fanout')
# 因为是广播的fanout模式，所以若不用接收广播则可以将queue删除
# 以下就是创建了一个exclusive[排他的，独立的queue
# 这个queue会在client上的channel断开后，自动被rabbixmq清除
result = channel.queue_declare("", exclusive=True)
queue_name = result.method.queue # 获取自动生成的queue的名字
print(queue_name)
# exchange不与consume直接关联，而是与queue关联，queue又与comsume的channel关联；
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(queue=queue_name, on_message_callback=callback, auto_ack=True)
channel.start_consuming()
Exchange的direct类型
关键字广播，consume连接队列，队列绑定关键字，发送者将数据与关键字发送到消息exchange，exchange根据关键字判定应该将数据发送至指定队列。
Producer端：
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
# 新建exchange名字为direct_log，exchange_type为'direct'
channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
# 需要一个关键字，即severity及消息即message
severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
# 消息发送，绑定exchange，routing_key指定关键字，body绑定message
channel.basic_publish(exchange='direct_logs', routing_key=severity, body=message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()Consumer端
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
# 新建exchange名字为direct_log，exchange_type为'direct'
channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
# 生成独占的自生自灭队列，并获取其名字
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue
# 获取需要收取的关键字信息，可以是多个
severities = sys.argv[1:]
if not severities:
sys.stderr.write("Usage: %s [info] [warning] [error]
" % sys.argv[0])
sys.exit(1)
# 将本consume的queue与exchange绑定，并绑定特定关键字，可重复绑定多个关键字
for severity in severities:
channel.queue_bind(exchange='direct_logs', 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(
queue=queue_name, on_message_callback=callback, auto_ack=True)
channel.start_consuming()Exchange的topic类型
topic[话题]类似于direct的关键字，区别在于topic拥有*及#
1.[*](star) can substitute for exactly one word.
2.[#](hash) can substitute for zero or more words.
3.单词间以[.]做分割标志.
Producer端：
#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs', exchange_type='topic')
routing_key = sys.argv[1] if len(sys.argv) > 2 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()Consumer端：
#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs', exchange_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]...
" % 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')
def callback(ch, method, properties, body):
print(" [x] %r:%r" % (method.routing_key, body))
channel.basic_consume(
queue=queue_name, on_message_callback=callback, auto_ack=True)
channel.start_consuming()RPC(Remote Procedure Call)
Our RPC will work like this:
- When the Client starts up, it creates an anonymous exclusive callback queue.
- For an RPC request, the Client sends a message with two properties: reply_to, which is set to the callback queue and correlation_id, which is set to a unique value for every request.
- The request is sent to an rpc_queue queue.
- The RPC worker (aka: server) is waiting for requests on that queue. When a request appears, it does the job and sends a message with the result back to the Client, using the queue from the reply_to field.
- The client waits for data on the callback queue. When a message appears, it checks the correlation_id property. If it matches the value from the request it returns the response to the application.
Client端
#!/usr/bin/env python
import pika
import uuid
class FibonacciRpcClient(object):
def __init__(self):
self.connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
self.channel = self.connection.channel()
# 新建一个自生自灭的独占队列，然后获取名字
result = self.channel.queue_declare(queue='', exclusive=True)
self.callback_queue = result.method.queue
# 声明我要收，使用上面的自生自灭独占队列收消息
self.channel.basic_consume(
queue=self.callback_queue,
on_message_callback=self.on_response, # 只要一收到消息就调用on_response
auto_ack=True)
def on_response(self, ch, method, props, body):
# 返回的props.correlation_id 与 self.corr_id对比，用于
# 确定是这个是我发的命令的执行结果
if self.corr_id == props.correlation_id:
self.response = body
def call(self, n):
self.response = None
self.corr_id = str(uuid.uuid4()) # 生成一个唯一的字符串
# 向rpc_queue发消息
self.channel.basic_publish(
exchange='',
routing_key='rpc_queue',
properties=pika.BasicProperties(
reply_to=self.callback_queue, # 告诉server你返回哪个队列，props是什么？
correlation_id=self.corr_id, # 发给了服务端->Server端
),
body=str(n)) # 发数字
while self.response is None:
# 这里面没有start_consuming->阻塞的
self.connection.process_data_events() # 这个是不阻塞的start_consuming
return int(self.response)
fibonacci_rpc = FibonacciRpcClient()
print(" [x] Requesting fib(30)")
response = fibonacci_rpc.call(30)
print(" [.] Got %r" % response)
Server端
#!/usr/bin/env python
import pika
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
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='',
routing_key=props.reply_to, # 获取client传过来的callback-queue
# correlation_id 返回.
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)
channel.basic_consume(queue='rpc_queue', on_message_callback=on_request)
print(" [x] Awaiting RPC requests")
channel.start_consuming()Memcached&Redis使用
rabbitMQ之类的队列解决了A程序与B程序间的数据传递，那么如果要实现数据的共享(读取同一块内存空间的数据)呢？
Broker-之缓存：
种类->mongodb<较老，默认持久化>,
redis<半持久化>
redis是一个key-value存储系统。和Memcached类似，它支持存储的value类型相对更多，包括string(字符串)、list(链表)、set(集合)、zset(sorted set --有序集合)和hash（哈希类型）。这些数据类型都支持push/pop、add/remove及取交集并集和差集及更丰富的操作，而且这些操作都是原子性的。在此基础上，redis支持各种不同方式的排序。与memcached一样，为了保证效率，数据都是缓存在内存中。区别的是redis会周期性的把更新的数据写入磁盘或者把修改操作写入追加的记录文件，并且在此基础上实现了master-slave(主从)同步。
具体介绍:
1. 使用Redis有哪些好处？ (1) 速度快，因为数据存在内存中，类似于HashMap，HashMap的优势就是查找和操作的时间复杂度都是O(1) (2) 支持丰富数据类型，支持string，list，set，sorted set，hash (3) 支持事务，操作都是原子性，所谓的原子性就是对数据的更改要么全部执行，要么全部不执行 (4) 丰富的特性：可用于缓存，消息，按key设置过期时间，过期后将会自动删除 2. redis相比memcached有哪些优势？ (1) memcached所有的值均是简单的字符串，redis作为其替代者，支持更为丰富的数据类型 (2) redis的速度比memcached快很多 (3) redis可以持久化其数据 3. redis常见性能问题和解决方案： (1) Master最好不要做任何持久化工作，如RDB内存快照和AOF日志文件 (2) 如果数据比较重要，某个Slave开启AOF备份数据，策略设置为每秒同步一次 (3) 为了主从复制的速度和连接的稳定性，Master和Slave最好在同一个局域网内 (4) 尽量避免在压力很大的主库上增加从库 (5) 主从复制不要用图状结构，用单向链表结构更为稳定，即：Master <- Slave1 <- Slave2 <- Slave3... 这样的结构方便解决单点故障问题，实现Slave对Master的替换。如果Master挂了，可以立刻启用Slave1做Master，其他不变。 4. MySQL里有2000w数据，redis中只存20w的数据，如何保证redis中的数据都是热点数据 相关知识：redis 内存数据集大小上升到一定大小的时候，就会施行数据淘汰策略。redis 提供 6种数据淘汰策略： voltile-lru：从已设置过期时间的数据集（server.db[i].expires）中挑选最近最少使用的数据淘汰 volatile-ttl：从已设置过期时间的数据集（server.db[i].expires）中挑选将要过期的数据淘汰 volatile-random：从已设置过期时间的数据集（server.db[i].expires）中任意选择数据淘汰 allkeys-lru：从数据集（server.db[i].dict）中挑选最近最少使用的数据淘汰 allkeys-random：从数据集（server.db[i].dict）中任意选择数据淘汰 no-enviction（驱逐）：禁止驱逐数据 5. Memcache与Redis的区别都有哪些？ 1)、存储方式 Memecache把数据全部存在内存之中，断电后会挂掉，数据不能超过内存大小。 Redis有部份存在硬盘上，这样能保证数据的持久性。 2)、数据支持类型 Memcache对数据类型支持相对简单。 Redis有复杂的数据类型。 3），value大小 redis最大可以达到1GB，而memcache只有1MB 6. Redis 常见的性能问题都有哪些？如何解决？ 1).Master写内存快照，save命令调度rdbSave函数，会阻塞主线程的工作，当快照比较大时对性能影响是非常大的，会间断性暂停服务，所以Master最好不要写内存快照。 2).Master AOF持久化，如果不重写AOF文件，这个持久化方式对性能的影响是最小的，但是AOF文件会不断增大，AOF文件过大会影响Master重启的恢复速度。Master最好不要做任何持久化工作，包括内存快照和AOF日志文件，特别是不要启用内存快照做持久化,如果数据比较关键，某个Slave开启AOF备份数据，策略为每秒同步一次。 3).Master调用BGREWRITEAOF重写AOF文件，AOF在重写的时候会占大量的CPU和内存资源，导致服务load过高，出现短暂服务暂停现象。 4). Redis主从复制的性能问题，为了主从复制的速度和连接的稳定性，Slave和Master最好在同一个局域网内 7, redis 最适合的场景 Redis最适合所有数据in-momory的场景，虽然Redis也提供持久化功能，但实际更多的是一个disk-backed的功能，跟传统意义上的持久化有比较大的差别，那么可能大家就会有疑问，似乎Redis更像一个加强版的Memcached，那么何时使用Memcached,何时使用Redis呢? 如果简单地比较Redis与Memcached的区别，大多数都会得到以下观点： 、Redis不仅仅支持简单的k/v类型的数据，同时还提供list，set，zset，hash等数据结构的存储。 、Redis支持数据的备份，即master-slave模式的数据备份。 、Redis支持数据的持久化，可以将内存中的数据保持在磁盘中，重启的时候可以再次加载进行使用。 （1）、会话缓存（Session Cache） 最常用的一种使用Redis的情景是会话缓存（session cache）。用Redis缓存会话比其他存储（如Memcached）的优势在于：Redis提供持久化。当维护一个不是严格要求一致性的缓存时，如果用户的购物车信息全部丢失，大部分人都会不高兴的，现在，他们还会这样吗？ 幸运的是，随着 Redis 这些年的改进，很容易找到怎么恰当的使用Redis来缓存会话的文档。甚至广为人知的商业平台Magento也提供Redis的插件。 （2）、全页缓存（FPC） 除基本的会话token之外，Redis还提供很简便的FPC平台。回到一致性问题，即使重启了Redis实例，因为有磁盘的持久化，用户也不会看到页面加载速度的下降，这是一个极大改进，类似PHP本地FPC。 再次以Magento为例，Magento提供一个插件来使用Redis作为全页缓存后端。 此外，对WordPress的用户来说，Pantheon有一个非常好的插件 wp-redis，这个插件能帮助你以最快速度加载你曾浏览过的页面。 （3）、队列 Reids在内存存储引擎领域的一大优点是提供 list 和 set 操作，这使得Redis能作为一个很好的消息队列平台来使用。Redis作为队列使用的操作，就类似于本地程序语言（如Python）对 list 的 push/pop 操作。 如果你快速的在Google中搜索“Redis queues”，你马上就能找到大量的开源项目，这些项目的目的就是利用Redis创建非常好的后端工具，以满足各种队列需求。例如，Celery有一个后台就是使用Redis作为broker，你可以从这里去查看。 （4），排行榜/计数器 Redis在内存中对数字进行递增或递减的操作实现的非常好。集合（Set）和有序集合（Sorted Set）也使得我们在执行这些操作的时候变的非常简单，Redis只是正好提供了这两种数据结构。所以，我们要从排序集合中获取到排名最靠前的10个用户–我们称之为“user_scores”，我们只需要像下面一样执行即可： 当然，这是假定你是根据你用户的分数做递增的排序。如果你想返回用户及用户的分数，你需要这样执行：ZRANGE user_scores 0 10 WITHSCORES Agora Games就是一个很好的例子，用Ruby实现的，它的排行榜就是使用Redis来存储数据的，你可以在这里看到。 （5）、发布/订阅 最后（但肯定不是最不重要的）是Redis的发布/订阅功能。发布/订阅的使用场景确实非常多。我已看见人们在社交网络连接中使用，还可作为基于发布/订阅的脚本触发器，甚至用Redis的发布/订阅功能来建立聊天系统！（不，这是真的，你可以去核实）。 Redis提供的所有特性中，我感觉这个是喜欢的人最少的一个，虽然它为用户提供如果此多功能。
memcache<不可持久化，轻量级缓存>
memcache是一个高性能的分布式内存对象缓存系统，用于动态Web应用以减轻数据库负载。它通过在内存中缓存数据和对象来减少读取数据库的次数，从而提高动态、数据库驱动网站的速度。Memcached基于一个存储键/值对的hashmap。其守护进程（daemon ）是用C写的，但是客户端可以用任何语言来编写，并通过memcached协议与守护进程通信。
Redis Python API使用：
redis-py 的API的使用可以分类为：
- 连接方式-最简单的连接及连接池
- 操作-String/Hash/List/Set/Sor Set/其他
- 其他功能-管道/发布订阅/Sentinel<哨兵模式>
连接方式-最简单的连接
import redis
r = redis.Redis(host="192.168.0.106", port=6379)
r.set('name', 'Bruce')
print(r.get("foo"))连接方式-连接池
redis-py使用connection pool来管理对一个redis server的所有连接，避免每次建立、释放连接的开销。默认，每个Redis实例都会维护一个自己的连接池。可以直接建立一个连接池，然后作为参数Redis，这样就可以实现多个Redis实例共享一个连接池。
import redis
pool = redis.ConnectionPool(host="192.168.0.106", port=6379)
r = redis.Redis(connection_pool=pool)
r.set('name', 'Bruce')
print(r.get("name"))操作-String
redis中的String在在内存中按照一个name对应一个value来存储。如图：
set(name, value, ex=None, px=None, nx=False, xx=False) 在Redis中设置值
在Redis中设置值，默认，不存在则创建，存在则修改
参数：
ex，过期时间（秒）
px，过期时间（毫秒）
nx，如果设置为True，则只有name不存在时，当前set操作才执行
xx，如果设置为True，则只有name存在时，岗前set操作才执行setnx(name, value) 同 set(name, value nx=True)
setex(name, value, time) 过期时间s
psetex(name, time_ms, value) 过期时间ms
mset(*args, **kwargs) 批量设置值，参数为字典或关键字
get(name) 获取值
mget(keys, *args) 批量获取值，参数为list或keys
getset(name, value) 获取之前的值同时赋予新值
getrange(key, start, end) string切片式获取按照字节切片
setrange(name, offset, value) 从offset处开始向后覆盖
setbit(name, offset, value) 将offset处的bit值替换为vlaue
getbit(name, offset) 获取offset处的bit值
bitcount(key, start=None, end=None) 获取key中二进制1的个数
bitop(operation, dest, *keys) 将keys进行位运算，并将值存入dest, 操作有 AND OR NOT XOR
strlen(key) 返回key值的字节长度
incr(self, name, amount=1) 自增amount的值
decr(self, name, amount=1) 自减amount的值
incrbyfloat(self, name, amount=1.0) # 自增amount的值，可以为浮点数
setbitgetbititcount与保存1/0数据的应用场景
# 场景如下，如果一个微博有1亿个用户的实时在线存储：
# 1.使用mysql存储用户oid对象的status变更为1，需要几百兆的空间，同时mysql查询速度会变慢；
# 2.使用setbit, getbit, bitcount这三个方法来实现；
##具体实现：
# 1.如果oid为1000的用户上线则，就setbit login_status 1000 1；
# 2.如果查询所有的在线人数，使用bitcount login_status 即可获取状态；
# 3.如果要查询某个用户是否在线，只要使用getbit login_status 是否为1即可判断状态；
import redis
import random
pool = redis.ConnectionPool(host='192.168.0.106', port=6379) # 建立连接池
r = redis.Redis(connection_pool=pool) # 建立redis对象
r.set("login_status", 0)
# 随机为10万用户分配登录状态
for oid in range(100000):
r.setbit("login_status", oid, random.randint(0, 1))
# 读取总在线人数
print("online counts in 100 thousand users>> ", r.bitcount("login_status"))
# 判断某个用户是否在线
user1_oid = random.randint(0, 100000)
user2_oid = random.randint(0, 100000)
user3_oid = random.randint(0, 100000)
user4_oid = random.randint(0, 100000)
print("oid [%s]'s status is %s." % (user1_oid, r.getbit("login_status", user1_oid)))))
print("oid [%s]'s status is %s." % (user2_oid, r.getbit("login_status", user2_oid)))
print("oid [%s]'s status is %s." % (user3_oid, r.getbit("login_status", user3_oid)))
print("oid [%s]'s status is %s." % (user4_oid, r.getbit("login_status", user4_oid)))
# 占用空间
print("100 thousand users online status take [%s] bytes" % r.strlen("login_status"))操作-Hash
Hash操作，redis中Hash在内存中的存储格式如下图：
hset(name, key, value) name对应的hash中设置一个键值对（不存在，则创建；否则，修改）
hmset(name, mapping) 在name对应的hash中批量设置键值对，map为dict
hget(name,key) 在name对应的hash中获取根据key获取value
hmget(name, keys, *args) 在那么对应的hash中获取多key对应的值
# 在name对应的hash中获取多个key的值
# 参数：
# name，reids对应的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对应hash的所有键值
hvals(name) 取name对应hash的所有值
hexists(name, key) 检查name对应的hash是否存在当前传入的key
hdel(name,*keys) 将name对应的hash中指定key的键值对删除
hincrby(name, key, amount=1) 自增name对应的hash中的指定key的值，不存在则创建key=amount
hincrbyfloat(name, key, amount=1.0) 自增name对应的hash中的指定key的值，不存在则创建key=amount
hscan(name, cursor=0, match=None, count=None) 增量式迭代获取，对于数据大的数据非常有用，hscan可以实现分片的获取数据，并非一次性将数据全部获取完，从而放置内存被撑爆
# 增量式迭代获取，对于数据大的数据非常有用，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封装hscan创建生成器，实现分批去redis中获取数据
# 参数：
# match，匹配指定key，默认None 表示所有的key
# count，每次分片最少获取个数，默认None表示采用Redis的默认分片个数
# 如：
# for item in r.hscan_iter('xx'):
# print item操作-List
redis中的List在在内存中按照一个name对应一个List来存储。如图：
lpush(name,values) 在name对应的list中添加元素，每个新的元素都添加到列表的最左边,rpush
lpushx(name,value) 在name对应的list中添加元素，只有name已经存在时，值添加到列表的最左边,rpushx
llen(name) name对应的list元素的个数
linsert(name, where, refvalue, value)) 在name对应的列表的某一个值前或后插入一个新值
lset(name, index, value) 对name对应的list中的某一个索引位置重新赋值
lrem(name, value, num) 在name对应的list中删除指定的值
lpop(name) 在name对应的列表的左侧获取第一个元素并在列表中移除，返回值则是第一个元素,rpop
lindex(name, index) 在name对应的列表中根据索引获取列表元素
lrange(name, start, end) 在name对应的列表分片获取数据
ltrim(name, start, end) 在name对应的列表中移除没有在start-end索引之间的值
rpoplpush(src, dst) 从一个列表取出最右边的元素，同时将其添加至另一个列表的最左边
blpop(keys, timeout) 按照从左到右去依次pop对应列表的元素，如果没有值等到超时，rlpop
brpoplpush(src, dst, timeout=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操作-Set
sadd(name,values) name对应的集合中添加元素
scard(name) 获取name对应的集合中元素个数
sdiff(keys, *args) 获取差集
sdiffstore(dest, keys, *args) 获取第一个name对应的集合中且不在其他name对应的集合，再将其新加入到dest对应的集合
sinter(keys, *args) 交集
sinterstore(dest, keys, *args) 交集
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) 并集并存储到dest
sscan(name, cursor=0, match=None, count=None)
sscan_iter(name, match=None, count=None) 同字符串的操作，用于增量迭代分批获取元素，避免内存消耗太大
操作-有序Set
有序集合，在集合的基础上，为每元素排序；元素的排序需要根据另外一个值来进行比较，所以，对于有序集合，每一个元素有两个值，即：值和分数，分数专门用来做排序。
zadd(name, *args, **kwargs) 在name对应的有序集合中添加元素，zadd name 15 “bb”
zcard(name) 获取name对应的有序集合元素的数量
zcount(name, min, max) 获取name对应的有序集合中分数 在 [min,max] 之间的个数,获取50-80分间的个数
zincrby(name, value, amount) 自增name对应的有序集合的 name 对应的分数
zrange( name, start, end, desc=False, withscores=False, score_cast_func=float) 按照索引范围获取name对应的有序集合的元素
# 参数：
# name，redis的name
# start，有序集合索引起始位置（非分数）
# 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 开始）
zrangebylex(name, min, max, start=None, num=None)
# 当有序集合的所有成员都具有相同的分值时，有序集合的元素会根据成员的 值 （lexicographical ordering）来进行排序，而这个命令则可以返回给定的有序集合键 key 中， 元素的值介于 min 和 max 之间的成员
# 对集合中的每个成员进行逐个字节的对比（byte-by-byte compare）， 并按照从低到高的顺序， 返回排序后的集合成员。 如果两个字符串有一部分内容是相同的话， 那么命令会认为较长的字符串比较短的字符串要大
# 参数：
# name，redis的name
# min，左区间（值）。 + 表示正无限； - 表示负无限； ( 表示开区间； [ 则表示闭区间
# min，右区间（值）
# start，对结果进行分片处理，索引位置
# num，对结果进行分片处理，索引后面的num个元素
# 如：
# ZADD myzset 0 aa 0 ba 0 ca 0 da 0 ea 0 fa 0 ga
# r.zrangebylex('myzset', "-", "[ca") 结果为：['aa', 'ba', 'ca']
# 更多：
# 从大到小排序
# zrevrangebylex(name, max, min, start=None, num=None)zrem(name, values) 删除name对应的有序集合中值是values的成员
zremrangebyrank(name, min, max) 根据排行范围删除
zremrangebyscore(name, min, max) 根据分数范围删除
zremrangebylex(name, min, max) 根据值返回删除
zscore(name, value) 获取name对应有序集合中 value 对应的分数
zinterstore(dest, keys, aggregate=None) 获取两个有序集合的交集，如果遇到相同值，则按照aggregate(sum min max)进行操作
zunionstore(dest, keys, aggregate=None) 获取两个有序集合的并集，如果遇到相同值，则按照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)
操作-其他
delete(*names) 根据删除redis中的任意数据类型
exists(name) 检测redis的name是否存在
keys(pattern='*') 根据模型获取redis的name
# 根据模型获取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下，redis默认有16个db使用select进行切换；
randomkey() 随机获取一个redis的name（不删除）
type(name) 获取name对应值的类型
scan(cursor=0, match=None, count=None)
scan_iter(match=None, count=None)
其他功能-管道
redis-py默认在执行每次请求都会创建（连接池申请连接）和断开（归还连接池）一次连接操作，如果想要在一次请求中指定多个命令，则可以使用pipline实现一次请求指定多个命令，并且默认情况下一次pipline 是原子性操作。
import redis, time
pool = redis.ConnectionPool(host='192.168.0.106', port=6379)
r = redis.Redis(connection_pool=pool)
# pipe = r.pipeline(transaction=False)
pipe = r.pipeline(transaction=True)
pipe.set('name123', 'Bruce') # 这个时候并不执行
time.sleep(50)
pipe.set('Luck11', 'Yes') # 这个时候并不执行
pipe.execute() # 统一执行，在sleep的50s中，没有执行，直到这条命令才执行其他功能-发布与订阅
发布者：服务器
订阅者：Dashboad和数据处理
Demo如下:
import redis
class RedisHelper:
def __init__(self):
self.__conn = redis.Redis(host='192.168.0.106')
self.chan_sub = 'fm104.5'
self.chan_pub = 'fm104.5'
def public(self, msg):
self.__conn.publish(self.chan_pub, msg) # 发布，调用Redis的publish方法即可
return True
def subscribe(self):
pub = self.__conn.pubsub() # 相当于打开收音机
pub.subscribe(self.chan_sub) # 调频道
pub.parse_response() # 准备接收
return pub
发布者:
from monitor.RedisHelper import RedisHelper
obj = RedisHelper()
obj.public('hello')订阅者:
from monitor.RedisHelper import RedisHelper
obj = RedisHelper()
redis_sub = obj.subscribe()
while True:
msg= redis_sub.parse_response() # 收到就继续，收不到就卡住
print(msg)更多Redis操作，参考http://doc.redisfans.com/
其他功能-Sentinel<哨兵>
主从切换技术的方法是：当主服务器宕机后，需要手动把一台从服务器切换为主服务器，这就需要人工干预，费事费力，还会造成一段时间内服务不可用。这不是一种推荐的方式，更多时候，我们优先考虑哨兵模式。
哨兵模式是一种特殊的模式，首先Redis提供了哨兵的命令，哨兵是一个独立的进程，作为进程，它会独立运行。其原理是哨兵通过发送命令，等待Redis服务器响应，从而监控运行的多个Redis实例。
这里的哨兵有两个作用
1.通过发送命令，让Redis服务器返回监控其运行状态，包括主服务器和从服务器。
2.当哨兵监测到master宕机，会自动将slave切换成master，然后通过发布订阅模式通知其他的从服务器，修改配置文件，让它们切换主机。注:然而一个哨兵进程对Redis服务器进行监控，可能会出现问题，为此，我们可以使用多个哨兵进行监控。各个哨兵之间还会进行监控，这样就形成了多哨兵模式，详细请戳这里
Python内的Sentinel的操作
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from redis.sentinel import Sentinel
# 连接哨兵服务器(主机名也可以用域名)
sentinel = Sentinel([('10.211.55.20', 26379),
('10.211.55.20', 26380),
],
socket_timeout=0.5)
# # 获取主服务器地址
# master = sentinel.discover_master('mymaster')
# print(master)
#
# # # 获取从服务器地址
# slave = sentinel.discover_slaves('mymaster')
# print(slave)
#
#
# # # 获取主服务器进行写入
# master = sentinel.master_for('mymaster')
# master.set('foo', 'bar')
# # # # 获取从服务器进行读取（默认是round-roubin）
# slave = sentinel.slave_for('mymaster', password='redis_auth_pass')
# r_ret = slave.get('foo')
# print(r_ret)TASK:
可以异步的执行多个命令，同时对多台机器
>>: run “df –h ” –hosts 192.168.3.55 10.4.3.4
task id: 12
>>: check_task 12
>>:
半成品-Client:
import pika
import uuid
class CmdRpcClient(object):
def __init__(self, remote_server, mq_Server="localhost"):
self.remote_server = remote_server
self.result = dict()
self.connection = pika.BlockingConnection(
pika.ConnectionParameters(host=mq_Server)
)
self.channel = self.connection.channel()
result = self.channel.queue_declare(queue='', exclusive=True)
self.callback_queue = result.method.queue
# 收取回复的消息的
self.channel.basic_consume(queue=self.callback_queue,
on_message_callback=self.on_response,
auto_ack=False)
# 处理收到的回复
def on_response(self, ch, method, properties, body):
for cmd in self.result:
if self.result[cmd]["CorrId"] == properties.correlation_id:
self.result[cmd]["result"] = body.decode()
self.result[cmd]["complete"] = True
break
ch.basic_ack(delivery_tag=method.delivery_tag) # 确认消息
# 向外发命令
def call(self, cmd):
corr_id = str(uuid.uuid4()) # 生成一个唯一的字符串
self.result[cmd] = dict()
self.result[cmd]["CorrId"] = corr_id
self.result[cmd]["complete"] = False
self.channel.queue_declare(queue=self.remote_server)
self.channel.basic_publish(exchange='',
routing_key=self.remote_server,
properties=pika.BasicProperties(
reply_to=self.callback_queue,
correlation_id=corr_id
),
body=cmd)
def check(self):
check_over = True
for cmd in self.result:
if not self.result[cmd]["complete"]:
check_over = False
self.connection.process_data_events()
return check_over
def get_result(self, cmd):
if self.result[cmd]["complete"]:
print(self.result[cmd]["result"])
del self.result[cmd]
else:
print("no result of cmd [{}]:
oid: [{}]".format(
cmd, self.result[cmd]["CorrId"]))
def results(self):
self.check()
for cmd in self.result:
print(cmd, "->", self.result[cmd]["complete"])
get_who = input("cmd>>")
if get_who in self.result:
self.get_result(get_who)
c1 = CmdRpcClient("server1")
print("1-run cmd
2-get result
3-quit
--------------------")
while True:
msg = input(">>:")
if not msg:
continue
elif msg == "1":
cmd = input("CMD>>:")
print(cmd)
c1.call(cmd)
elif msg == "2":
c1.results()
elif msg == "quit":
break
半成品-Server:
import pika
import gevent
from gevent import monkey
import random
import time
import os
class CmdRpcServer(object):
def __init__(self, my_host, mq_Server="localhost"):
self.host = my_host
self.connection = pika.BlockingConnection(
pika.ConnectionParameters(host=mq_Server))
self.channel = self.connection.channel()
self.channel.queue_declare(queue=self.host)
def runCmd(self, ch, method, props, body):
result = os.popen(body.decode()).read()
time.sleep(random.randint(1, 10))
print("the cmd [{}]'s result is [{}].".format(body, result))
ch.basic_publish(exchange='', routing_key=props.reply_to,
properties=pika.BasicProperties(
correlation_id=props.correlation_id), body=result)
ch.basic_ack(delivery_tag=method.delivery_tag)
def on_request(self, ch, method, props, body):
monkey.patch_all()
gevent.spawn(self.runCmd, ch, method, props, body)
def on_recv(self):
self.channel.basic_consume(queue=self.host,
on_message_callback=self.on_request)
print(" [x] Awaiting RPC requests")
self.channel.start_consuming()
server1 = CmdRpcServer("server1")
server1.on_recv()
end
官网文档-https://www.rabbitmq.com/getstarted.html
Membercache&Redis-https://www.cnblogs.com/wupeiqi/articles/5132791.html
ubuntu安装-https://www.cnblogs.com/yunlongaimeng/p/10260573.html
Redis的Sentinel模式-https://www.jianshu.com/p/06ab9daf921d