使用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会重新将该任务添加到队列中