redis补充和rabbitmq讲解

fairyguo

　　线程池；
　　redis发布订阅；
　　rabbitMQ；
　　MySQL;
　　python pymysql；
　　python orm SQLAchemy；
　　paramiko；
　　堡垒机；

1.线程池

1.1 contextlib ，上下文切换



import contextlib
#这个装饰器赋予work_state这个方法能使用“with work_state”，实现了一个上下文切换的功能
@contextlib.contextmanager
def work_state(state_list,work_thread):  #①
state_list.append(work_thread)           #②
try:
yield                                               #③
finally:
state_list.remove(work_thread)       #⑨

state_list = []
work_thread = 'fdafd'
with work_state(state_list,work_thread): #④
print(123)                                           #⑤
print(state_list)                                   #⑥
print(work_thread)                              #⑦
print(456)                                           #⑧
#结果：
123
['fdafd']
fdafd
456


import contextlib
import socket
@contextlib.contextmanager
def context_sock(host,port):
sk = socket.socket()
sk.bind((host,port))
sk.listen(5)
try:
yield sk    #将sk传给下面的“with context_sock...as sock”
finally:
sk.close()
with context_sock('localhost',8888) as sock:
print(sock)   #此处的sock就是sk。
2.redis 补充
　　redis-py使用connection pool来管理对一个redis server的所有连接，避免每次建立、释放连接的开销。默认，每个Redis实例都会维护一个自己的连接池。可以直接建立一个连接池，然后作为参数Redis，这样就可以实现多个Redis实例共享一个连接池。



#!/usr/bin/env python
# -*- coding:utf-8 -*-

import redis

#创建一个连接池，之后与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'))
2.1redis发布订阅



#redis_pub_sub.py
import redis
class RedisHelper(object):
def __init__(self):
self.__conn = redis.Redis(host='127.0.0.1')
def public(self,msg,channel):
self.__conn.publish(channel,msg)   #先channel后msg。
return True
def subscribe(self,channel):
pub = self.__conn.pubsub()
pub.subscribe(channel)
pub.parse_response()
return pub
#sub.py
import redis_pub_sub
sub1 = redis_pub_sub.RedisHelper()
data = sub1.subscribe('fm104.99')
print(data.parse_response())
#可以起多个sub，当pub发布了消息，这些sub都能收到。
#pub.py
import redis_pub_sub
pub1 = redis_pub_sub.RedisHelper()
pub1.public('avadv','fm104.99')
3. 初识rabbitMQ
　　rabbitmq实现发布订阅；pika是使用rabbitmq的插件。



#publisher
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')
channel.basic_publish(exchange='',
routing_key='hello',
body='first')
print('publish ok')
connection.close()

#consumer
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')
def callable(ch,method,properties,body):
print("[x] Received %r" % body)
channel.basic_consume(callable,
queue='hello',
no_ack=True)
print('

Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
3.1 ackownledgment，消息不丢失
　　上面的代码实现了发布订阅，但是有个问题，假如consumer在接收消息的过程中宕了，就会出现consumer并没有接受完消息，但是消息队列里也没消息了，即使consumer恢复了也没法再重新获取到消息了，这时候就需要设置一下消息不丢失。
　　消息不丢失只需要修改consumer端的代码即可，其实是改成“consumer接受完消息后，会给消息队列一个接收完毕的消息，然后这条消息就从队列里移除了，如果consumer没有发确认消息，此条消息就一直存在消息队列里”。



import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')
def callable(ch,method,properties,body):
print("[x] Received %r" % body)
import time
time.sleep(50)
print('consumer ok')
ch.basic_ack(delivery_tag = method.delivery_tag) #接收完消息后，给消息队列发确认收到的消息。

channel.basic_consume(callable,
queue='hello',
no_ack=False      #这一行表示需要消息确认；no_ack的全称是no ackownledge。
                      )
print('

Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
3.2 during,消息持久化
　　消息持久化是在publish端配置



#!/usr/bin/env python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4'))
channel = connection.channel()
# make message persistent
channel.queue_declare(queue='hello', durable=True)
channel.basic_publish(exchange='',
routing_key='hello',
body='Hello World!',
properties=pika.BasicProperties(
delivery_mode=2, # 设置了这里，就使消息持久化了。
                      ))
print(" [x] Sent 'Hello World!'")
connection.close()
　　publish端配置了持久化，consumer端也配置了持久化，这样无论哪端宕了都不会丢失消息。

3.3 消息获取顺序
　　默认的话，各个consumer获取消息的顺序是轮询来取的，consumer1取消息1，consumer2取消息2，consumer3取消息3，这时候如果consumer3先处理完了，需要再取消息，那consumer3取的是6，因为4和5之后会被consumer1和consumer2取走。
　　如果想做到哪个consumer先处理完了就接着取接下来的消息的话，即各个consumer顺序取消息，那就需要在consumer代码里加下述参数：channel.basic_qos(prefetch_count=1)
　　这样如果consumer3先处理完了，再取消息就是消息4了，而不是消息6。

3.4 发布和订阅

3.4.1 exchange
　　exchange是publisher和rabbitmq间的一个中间层，publisher给rabbitmq发消息，其实是先发给了exchange，然后exchange根据规则发到对应的消息队列里。
　　exchange的作用，比如你想给所有的消息队列发一条消息，此时rabbitmq里有100个消息队列，如果没有exchange的话，你需要自己发送100次消息，如果有了exchange，你需要做的仅仅是将消息发送给rabbitmq，然后选择一个对应的规则，比如说是广播，那exchange就会把消息发给所有绑定到exchange的队列。这里仅仅是举了一个广播的例子，exchange还有其他发布模式。

3.4.2 exchange的fanout模式，广播
　　publisher



#!/usr/bin/env python

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
#声明一个exchange，名称和类型
channel.exchange_declare(exchange='logs_exchange' , type='fanout')
message = 'hhh'
#向这个exchange发送消息
channel.basic_publish(exchange='logs_exchange',routing_key='',body=message)

print('publish ok')
connection.close()
　　consumer



#!/usr/bin/env python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
#声明一个exchange
channel.exchange_declare(exchange='logs_exchange',type='fanout')
#下面两行代码的作用是声明一个随机名字的消息队列，这样每次执行这些代码，就会生成不同的消息队列；如果指定了队列名字，那每次执行都是同一个队列。
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
#绑定exchange
channel.queue_bind(exchange='logs_exchange',queue=queue_name)
print('waiting the message coming...')
def callable(ch,method,properties,body):
print('recieve the %s' % body)
channel.basic_consume(callable,queue=queue_name,no_ack=True)
channel.start_consuming()
　　多次执行consumer的代码，生成了多个消息队列，所有的consumer都会同时收到publisher的消息。

3.4.3 exchange的direct模式，基于关键字来发送消息
　　发送消息时明确指定某个队列并向其中发送消息，RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字 判定应该将数据发送至指定队列。
　　publisher指定将消息发给绑定了info这个关键字的队列；consumer1绑定了info、error关键字；consumer2绑定了error关键字；则，只有consumer1能收到publisher的消息。如果publisher将消息发给绑定了error关键字的队列，则consumer1和consumer2都能收到消息。这就是基于关键字来管理消息。
　　生产者



#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',
type='direct')
#输入一个关键字
severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
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()
　　消费者



#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',
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]\n" % sys.argv[0])
sys.exit(1)
#循环绑定关键字；routing_key是变化的
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()
3.4.4 模糊匹配
　　在topic类型下，可以让队列绑定几个模糊的关键字，之后发送者将数据发送到exchange，exchange将传入”路由值“和 ”关键字“进行匹配，匹配成功，则将数据发送到指定队列。

• # 表示可以匹配 0 个 或 多个 单词
  
• *  表示只能匹配 一个 单词
  

发送者路由值              队列中
old.boy.python          old.*  -- 不匹配
old.boy.python          old.#  -- 匹配
　　saltstack支持模糊匹配，虽然saltstack用的zeromq，但是模糊匹配的原理跟这个类似。
　　生产者



#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs',
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()
　　消费者



#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
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')
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()