线程池;
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()