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

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