python之上下文管理、redis的发布订阅、rabbitmq

孤独750

　　使用with打开文件的方式，是调用了上下文管理的功能



1 #打开文件的两种方法：
2
3 f = open('a.txt','r')
4
5 with open('a.txt','r') as f
6
7 实现使用with关闭socket
8 import contextlib
9 import socket
10
11 @contextlib.contextmanage
12 def Sock(ip,port):
13     socket = socket.socket()
14     socket.bind((ip,port))
15     socket.listen(5)
16     try:
17         yield socket
18     finally:
19         socket.close()
20
21 #执行Sock函数传入参数，执行到yield socket返回值给s，执行with语句体，执行finally后面的语句
22 with Sock('127.0.0.1',8000) as s:
23     print(s)
redis的发布订阅



class RedisHelper:
def __init__(self):
#调用类时自动连接redis
self.__conn = redis.Redis(host='192.168.1.100')
def public(self, msg, chan):
self.__conn.publish(chan, msg)
return True
def subscribe(self, chan):
pub = self.__conn.pubsub()
pub.subscribe(chan)
pub.parse_response()
return pub
#订阅者
import s3
obj = s3.RedisHelper()
data = obj.subscribe('fm111.7')
print(data.parse_response())
#发布者
import s3
obj = s3.RedisHelper()
obj.public('alex db', 'fm111.7')
RabbitMQ



1 #消费者
2 import pika
3
4 connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))
5 channel = connection.channel()#创建对象
6
7 channel.queue_declare(queue = 'wocao')
8 def callback(ch,method,properties,body):
9     print("[x] Received %r"%body)
10
11 channel.basic_consume(callback,queue = 'wocao',no_ack = True)
12 print('

Waiting for messages. To exit press CTRL+C')
13 channel.start_consuming()
14
15 #生产者
16 import pika
17 connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))
18 channel = connection.channel()
19 channel.queue_declare(queue = 'wocao')#指定一个队列，不存在此队列则创建
20 channel.basic_publish(exchange = '',routing_key = 'wocao',body = 'hello world!')
21 print("[x] Sent 'hello world!")
22 connection.close()
exchange type类型



#生产者
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='192.168.11.87'))
channel = connection.channel()
#fanout类型，对绑定该exchange的队列实行广播
channel.exchange_declare(exchange='logs_fanout',
type='fanout')
# 随机创建队列
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
# 绑定exchange
channel.queue_bind(exchange='logs_fanout',
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()
#消费者
import pika
#发送方
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='192.168.11.87'))
channel = connection.channel()
channel.exchange_declare(exchange='logs_fanout',
type='fanout')
message = "what's the fuck"
#设置exchange的名
channel.basic_publish(exchange='logs_fanout',
routing_key='',
body=message)
print(" [x] Sent %r" % message)
connection.close()


1 #根据关键字发送指定队列
2 #生产者（发布者）
3 import pika
4 connection = pika.BlockingConnection(pika.ConnectionParameters(
5                                      host = '127.0.0.1'))
6 channel = connection.channel()
7
8 channel.exchange_declare(exchange='direct_logs_1',
9                          type='direct')  # 关键字发送到队列
10 #对error关键字队列发送指令
11 severity = 'error'
12 message = '123'
13 channel.basic_publish(exchange = 'direct_logs_1',
14                        routing_key = severity,
15                        body = message)
16 print('[x] Sent %r:%r'%(severity,message))
17 connection.close()
18 #消费者（订阅者）
19 import pika
20 #消费者
21 connection = pika.BlockingConnection(pika.ConnectionParameters(
22                                      host = '127.0.0.1'))
23 channel = connection.channel()
24 channel.exchange_declare(exchange='direct_logs_1',
25                          type = 'direct')#关键字发送到队列
26
27 result = channel.queue_declare(exclusive=True)
28 queue_name = result.method.queue
29 serverities = ['error','info','warning']
30 for severity in serverities:
31     channel.queue_bind(exchange='direct_logs_1',
32                        queue = queue_name,
33                        routing_key = severity)
34 def callback(ch,method,properties,body):
35     print('[x] %r:%r'%(method.routing_key,body))
36
37 channel.basic_consume(callback,
38                       queue = queue_name,
39                       no_ack = True)
40 channel.start_consuming()


1 #实现消息不丢失接收方
2 import pika
3 connection = pika.BlockingConnection(pika.ConnectionParameters(host = '10.211.55.4'))
4 channel = connection.channel()
5 channel.queue_declare(queue = 'hello')
6
7 def callback(ch,method,properties,body):
8     print('redeived %s'%body)
9     import time
10     time.sleep(10)
11     print('ok')
12     ch.basic_ack(delivery_tag= method.delivery_tag)
13 #no_ack = False接收方接受完请求后发送给对方一个接受成功的信号，如果没收到mq会重新将任务放到队列
14 channel.basic_consume(callback,queue = 'hello',no_ack=False)
15 print(' Waiting for messages.To exit press CTRL+C')
16 channel.start_consuming()


1 #发送方
2 #实现消息不丢失
3 import pika
4 connection = pika.BlockingConnection(pika.ConnectionParameters(host = '10.211.55.4'))
5 channel = connection.channel()
6 channel.queue_declare(queue = 'hello',durable = True)
7 channel.basic_publish(exchange = '',routing_key = 'hello world',
8                       properties = pika.BasicProperties(
9                           delivery_mode=2,
10                       ))#发送方不丢失，发送方保持持久化
11 print(' Waiting for messages.To exit press CTRL+C')
12 channel.start_consuming()


1 #接收方
2 import pika
3
4 connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.11.100'))
5 channel = connection.channel()
6
7
8 channel.queue_declare(queue='hello', durable=True)
9 def callback(ch, method, properties, body):
10     print(" [x] Received %r" % body)
11     import time
12     time.sleep(10)
13     print 'ok'
14     ch.basic_ack(delivery_tag = method.delivery_tag)
15 channel.basic_consume(callback,
16                       queue='hello',
17                       no_ack=False)
18 channel.start_consuming()
　　RabbitMQ队列中默认情况下，接收方从队列中获取消息是顺序的，例如：接收方1只从队列中获取奇数的任务，接收方2只从队列中获取偶数任务



import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.11.100'))
channel = connection.channel()
channel.queue_declare(queue='hello')
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
import time
time.sleep(10)
print 'ok'
ch.basic_ack(delivery_tag = method.delivery_tag)
#表示队列不分奇偶分配，谁来取任务就给谁
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
queue='hello',
no_ack=False)
print('

Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
　　RabbitMQ会重新将该任务添加到队列中