RabbitMQ队列
安装 RabbitMQ
RabbitMQ 使用Erlang语言开发的,需要先安装Erlang
1.先安装Erlang : http://www.erlang.org/downloads
2.再安装RabbitMQ : http://www.rabbitmq.com/install-windows.html
3.安装pika cmd下: C:UsersYP>pip install pika
消费者:
# !usr/bin/env python # -*-coding:utf-8 -*- import pika conncetion = pika.BlockingConnection(pika.ConnectionParameters("localhost")) channel = conncetion.channel() channel.queue_declare(queue="hello1",durable= True) #避免这个先启动,会报错,如果存在队列,这个会忽略 def callback(ch,method,properties,body): print("body",body) channel.basic_consume(callback, queue="hello1", #no_ack=True #默认无需写 ) print("CTRL + C") channel.start_consuming()
生产者:
# !usr/bin/env python # -*-coding:utf-8 -*- import pika connection = pika.BlockingConnection(pika.ConnectionParameters("localhost")) channel = connection.channel() #声明queue channel.queue_declare(queue = "hello1",durable= True) channel.basic_publish(exchange="", routing_key="hello1", body = "hello world!", ) print("[X] send 'hello world!'") connection.close()
远程连接rabbitmq server的话,需要配置权限
首先在rabbitmq server上创建一个用户
sudo
rabbitmqctl add_user alex alex3714
同时还要配置权限,允许从外面访问
sudo
rabbitmqctl set_permissions -p / alex
".*"
".*"
".*"
set_permissions [-p vhost] {user} {conf} {write} {read}
vhost:授予用户访问权限的虚拟主机的名称,默认为/.
user:授予对指定虚拟主机的访问权限的用户名
conf:一个正则表达式匹配资源名,用户被授予配置权限。
write:一个匹配资源名称的正则表达式,授予用户写权限。
read:一个正则表达式,匹配授予用户读权限的资源名称。
客户端连接的时候需要配置认证参数
credentials = pika.PlainCredentials('zyp', 'zyp') connection = pika.BlockingConnection(pika.ConnectionParameters( '10.211.55.5',5672,'/',credentials)) channel = connection.channel()
Work Queues
在这种模式下,RabbitMQ会默认把p发的消息依次分发给各个消费者(c),跟负载均衡差不多
生产者
import pika import time connection = pika.BlockingConnection(pika.ConnectionParameters( 'localhost')) channel = connection.channel() # 声明queue channel.queue_declare(queue='task_queue') import sys 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()
消费者
#_*_coding:utf-8_*_ import pika, time connection = pika.BlockingConnection(pika.ConnectionParameters( 'localhost')) channel = connection.channel() def callback(ch, method, properties, body): print(" [x] Received %r" % body) time.sleep(5) print("Done") print("method.delivery_tag",method.delivery_tag) ch.basic_ack(delivery_tag=method.delivery_tag) #手动确认收到消息
channel.basic_consume(callback, queue='task_queue', no_ack=True )
print(' CTRL+C')
channel.start_consuming()
此时,先启动消息生产者,然后再分别启动3个消费者,通过生产者多发送几条消息,你会发现,这几条消息会被依次分配到各个消费者身上
ch.basic_ack(delivery_tag=method.delivery_tag)
如果使用者在没有发送ack的情况下死亡(通道关闭、连接关闭或TCP连接丢失),RabbitMQ将理解消息没有被完整地处理,并将它重新排队。如果同时有其他消费者在线,它将很快重新交付给另一个消费者。这样你就可以确保没有信息丢失,即使工人偶尔也会死亡。
消息持久化
当RabbitMQ退出或崩溃时,它将忘记队列和消息,除非您告诉它不要这样做。确保消息不丢失需要两件事情:我们需要将队列和消息标记为持久的。
1,我们需要确保RabbitMQ永远不会丢失队列。为了做到这一点,我们需要声明它是持久的:队列持久化
channel.queue_declare(queue='hello', durable=True)
这个queue_declare更改需要同时应用于生产者代码和消费者代码。
2,现在我们需要将我们的消息标记为持久性—通过提供一个值为2的delivery_mode属性。
channel.basic_publish(exchange='', routing_key="task_queue", body=message, properties=pika.BasicProperties( delivery_mode = 2, # 消息持久化 ))
消息公平分发
如果Rabbit只管按顺序把消息发到各个消费者身上,不考虑消费者负载的话,很可能出现,一个机器配置不高的消费者那里堆积了很多消息处理不完,同时配置高的消费者却一直很轻松。为解决此问题,可以在各个消费者端,配置perfetch=1,意思就是告诉RabbitMQ在我这个消费者当前消息还没处理完的时候就不要再给我发新消息了。
channel.basic_qos(prefetch_count=1)
带消息持久化+公平分发的完整代码
生产者
# !usr/bin/env python # -*-coding:utf-8 -*- import pika connection = pika.BlockingConnection(pika.ConnectionParameters("localhost")) channel = connection.channel() #声明queue channel.queue_declare(queue = "hello1",durable= True) #队列持久化 channel.basic_publish(exchange="", routing_key="hello1", body = "hello world!", properties = pika.BasicProperties(delivery_mode= 2 ) #消息持久化 ) print("[X] send 'hello world!'") connection.close()
消费者
# !usr/bin/env python # -*-coding:utf-8 -*- import pika import time conncetion = pika.BlockingConnection(pika.ConnectionParameters("localhost")) channel = conncetion.channel() channel.queue_declare(queue="hello1",durable= True) #避免这个先启动,会报错,如果存在队列,这个会忽略 def callback(ch,method,properties,body): print("body",body) time.sleep(2) ch.basic_ack(delivery_tag=method.delivery_tag) #手动确认 channel.basic_qos(prefetch_count=1) #实现消费者端处理不过来就发给其他消费者 channel.basic_consume(callback, queue="hello1", #no_ack=True #默认无需写 需要查资料确认这个功能 ) print("CTRL + C") channel.start_consuming()
PublishSubscribe(消息发布订阅)
之前的例子都基本都是1对1的消息发送和接收,即消息只能发送到指定的queue里,但有些时候你想让你的消息被所有的Queue收到,类似广播的效果,这时候就要用到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
生产者
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
# !usr/bin/env python # -*-coding:utf-8 -*- import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='logs', exchange_type='fanout') message = ' '.join(sys.argv[1:]) or "info: Hello World!" channel.basic_publish(exchange='logs', routing_key='', body=message) print(" [x] Sent %r" % message) connection.close()
消费者
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
# !usr/bin/env python # -*-coding:utf-8 -*- import pika connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='logs', exchange_type='fanout') result = channel.queue_declare(exclusive=True) # 不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除 queue_name = result.method.queue 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()
有选择的接收消息(exchange type=direct)
RabbitMQ还支持根据关键字发送,即:队列绑定关键字,发送者将数据根据关键字发送到消息exchange,exchange根据 关键字 判定应该将数据发送至指定队列。
生产者
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
# !usr/bin/env python # -*-coding:utf-8 -*- import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='direct_logs', exchange_type='direct') severity = 'error' #获取执行后面跟的参数 #sys.argv[1] if len(sys.argv) > 1 else message = ' '.join(sys.argv[2:]) or 'Hello World!' channel.basic_publish(exchange='direct_logs', routing_key=severity, body=message) print(" [x] Sent %r:%r" % (severity, message)) connection.close()
消费者
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
# !usr/bin/env python # -*-coding:utf-8 -*- import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='direct_logs', exchange_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] [warning] [error] " % sys.argv[0]) sys.exit(1) 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(callback, queue=queue_name, no_ack=True) channel.start_consuming()
更细致的消息过滤
虽然使用直接交换改进了我们的系统,但它仍然有局限性——它不能基于多个条件进行路由。
在我们的日志系统中,我们可能不仅希望订阅基于严重性的日志,而且还希望订阅发送日志的源。您可能从syslog unix工具了解这个概念,该工具基于严重性(info/warn/crit…)和工具(auth/cron/kern…)路由日志。
这将给我们带来很大的灵活性——我们可能希望只听来自“cron”的关键错误,也希望听来自“kern”的所有日志。
生产者
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
# !usr/bin/env python # -*-coding:utf-8 -*- 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) > 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()
消费者
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
# !usr/bin/env python # -*-coding:utf-8 -*- 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(callback, queue=queue_name, no_ack=True) channel.start_consuming()
Remote procedure call (RPC)
使用 RPC 编程是在分布式环境中运行的客户机和服务器应用程序之间进行可靠通信的最强大、最高效的方法之一。
实例:客户端输入要计算的斐波那契数列的值,通过服务端计算出值,并将结果返回给客户端
fibonacci_rpc
=
FibonacciRpcClient()
result
=
fibonacci_rpc.call(
4
)
print
(
"fib(4) is %r"
%
result)
![](https://images2018.cnblogs.com/blog/1019324/201806/1019324-20180624093959376-300816044.png)
RPC server
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
import pika import time 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, properties=pika.BasicProperties(correlation_id= props.correlation_id), body=str(response)) ch.basic_ack(delivery_tag=method.delivery_tag) channel.basic_consume(on_request, queue='rpc_queue') print(" [x] Awaiting RPC requests") channel.start_consuming()
RPC client
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
import pika import uuid import time class FibonacciRpcClient(object): def __init__(self): self.connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) self.channel = self.connection.channel() result = self.channel.queue_declare(exclusive=True) self.callback_queue = result.method.queue self.channel.basic_consume(self.on_response, #只要已收到消息就调用on_response no_ack=True, queue=self.callback_queue) def on_response(self, ch, method, props, body): 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='', routing_key='rpc_queue', properties=pika.BasicProperties( reply_to=self.callback_queue, correlation_id=self.corr_id, ), body=str(n)) while self.response is None: self.connection.process_data_events() #非阻塞版的start_consuming() time.sleep(0.5) return int(self.response) fibonacci_rpc = FibonacciRpcClient() print(" [x] Requesting fib(30)") response = fibonacci_rpc.call(6) print(" [.] Got %r" % response)
缓存数据库介绍
NoSQL(NoSQL = Not Only SQL ),意即“不仅仅是SQL”,泛指非关系型的数据库,随着互联网web2.0网站的兴起,传统的关系数据库在应付web2.0网站,特别是超大规模和高并发的SNS类型的web2.0纯动态网站已经显得力不从心,暴露了很多难以克服的问题,而非关系型的数据库则由于其本身的特点得到了非常迅速的发展。NoSQL数据库的产生就是为了解决大规模数据集合多重数据种类带来的挑战,尤其是大数据应用难题。
NoSQL数据库的四大分类
NoSQL数据库的四大分类表格分析
Redis使用
介绍
redis是业界主流的key-value nosql 数据库之一。和Memcached类似,它支持存储的value类型相对更多,包括string(字符串)、list(链表)、set(集合)、zset(sorted set --有序集合)和hash(哈希类型)。这些数据类型都支持push/pop、add/remove及取交集并集和差集及更丰富的操作,而且这些操作都是原子性的。在此基础上,redis支持各种不同方式的排序。与memcached一样,为了保证效率,数据都是缓存在内存中。区别的是redis会周期性的把更新的数据写入磁盘或者把修改操作写入追加的记录文件,并且在此基础上实现了master-slave(主从)同步。
Redis优点
-
异常快速 : Redis是非常快的,每秒可以执行大约110000设置操作,81000个/每秒的读取操作。
-
支持丰富的数据类型 : Redis支持最大多数开发人员已经知道如列表,集合,可排序集合,哈希等数据类型。
这使得在应用中很容易解决的各种问题,因为我们知道哪些问题处理使用哪种数据类型更好解决。 -
操作都是原子的 : 所有 Redis 的操作都是原子,从而确保当两个客户同时访问 Redis 服务器得到的是更新后的值(最新值)。
-
MultiUtility工具:Redis是一个多功能实用工具,可以在很多如:缓存,消息传递队列中使用(Redis原生支持发布/订阅),在应用程序中,如:Web应用程序会话,网站页面点击数等任何短暂的数据;
安装Redis环境
$sudo apt-get update $sudo apt-get install redis-server
启动 Redis
$redis-server
查看 redis 是否还在运行
$redis-cli
redis 127.0.0.1:6379>
redis 127.0.0.1:6379> ping PONG
Python操作Redis
在Ubuntu上安装Redis桌面管理器
要在Ubuntu 上安装 Redis桌面管理,可以从 http://redisdesktop.com/download 下载包并安装它。
Redis API使用
redis-py 的API的使用可以分类为:
- 连接方式
- 连接池
- 操作
- String 操作
- Hash 操作
- List 操作
- Set 操作
- Sort Set 操作
- 管道
- 发布订阅
连接方式
1、操作模式
redis-py提供两个类Redis和StrictRedis用于实现Redis的命令,StrictRedis用于实现大部分官方的命令,并使用官方的语法和命令,Redis是StrictRedis的子类,用于向后兼容旧版本的redis-py。
import redis r = redis.Redis(host='192.18.66.134', port=6379) r.set('foo', 'Bar') print r.get('foo')
2、连接池
redis-py使用connection pool来管理对一个redis server的所有连接,避免每次建立、释放连接的开销。默认,每个Redis实例都会维护一个自己的连接池。可以直接建立一个连接池,然后作为参数Redis,这样就可以实现多个Redis实例共享一个连接池。
1. String操作
redis中的String在在内存中按照一个name对应一个value来存储
set(name, value, ex=None, px=None, nx=False, xx=False)
在Redis中设置值,默认,不存在则创建,存在则修改
参数:
ex,过期时间(秒)
px,过期时间(毫秒)
nx,如果设置为True,则只有name不存在时,当前set操作才执行
xx,如果设置为True,则只有name存在时,岗前set操作才执行