使用with打开文件的方式，是调用了上下文管理的功能
#打开文件的两种方法：
f = open('a.txt','r')
with open('a.txt','r') as f
实现使用with关闭socket
import contextlib
import socket
@contextlib.contextmanage
def Sock(ip,port):
socket = socket.socket()
socket.bind((ip,port))
socket.listen(5)
try:
yield socket
finally:
socket.close()
#执行Sock函数传入参数，执行到yield socket返回值给s，执行with语句体，执行finally后面的语句
with Sock('127.0.0.1',8000) as s:
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
#消费者
import pika connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))
channel = connection.channel()#创建对象 channel.queue_declare(queue = 'wocao')
def callback(ch,method,properties,body):
print("[x] Received %r"%body) channel.basic_consume(callback,queue = 'wocao',no_ack = True)
print('[*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming() #生产者
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))
channel = connection.channel()
channel.queue_declare(queue = 'wocao')#指定一个队列，不存在此队列则创建
channel.basic_publish(exchange = '',routing_key = 'wocao',body = 'hello world!')
print("[x] Sent 'hello world!")
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()
#根据关键字发送指定队列
#生产者（发布者）
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
host = '127.0.0.1'))
channel = connection.channel() channel.exchange_declare(exchange='direct_logs_1',
type='direct') # 关键字发送到队列
#对error关键字队列发送指令
severity = 'error'
message = ''
channel.basic_publish(exchange = 'direct_logs_1',
routing_key = severity,
body = message)
print('[x] Sent %r:%r'%(severity,message))
connection.close()
#消费者（订阅者）
import pika
#消费者
connection = pika.BlockingConnection(pika.ConnectionParameters(
host = '127.0.0.1'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs_1',
type = 'direct')#关键字发送到队列 result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
serverities = ['error','info','warning']
for severity in serverities:
channel.queue_bind(exchange='direct_logs_1',
queue = queue_name,
routing_key = severity)
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()
#实现消息不丢失接收方
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host = '10.211.55.4'))
channel = connection.channel()
channel.queue_declare(queue = 'hello') def callback(ch,method,properties,body):
print('redeived %s'%body)
import time
time.sleep(10)
print('ok')
ch.basic_ack(delivery_tag= method.delivery_tag)
#no_ack = False接收方接受完请求后发送给对方一个接受成功的信号，如果没收到mq会重新将任务放到队列
channel.basic_consume(callback,queue = 'hello',no_ack=False)
print(' Waiting for messages.To exit press CTRL+C')
channel.start_consuming()
#发送方
#实现消息不丢失
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host = '10.211.55.4'))
channel = connection.channel()
channel.queue_declare(queue = 'hello',durable = True)
channel.basic_publish(exchange = '',routing_key = 'hello world',
properties = pika.BasicProperties(
delivery_mode=2,
))#发送方不丢失，发送方保持持久化
print(' Waiting for messages.To exit press CTRL+C')
channel.start_consuming()
#接收方
import pika connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.11.100'))
channel = connection.channel() channel.queue_declare(queue='hello', durable=True)
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_consume(callback,
queue='hello',
no_ack=False)
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会重新将该任务添加到队列中