继续在web开发的路上前行

61581229

　　之前学的通信，异步，现在回过头来看，已经没有那么难，希望以后能排上用场。这个星期学的rabbitmq，redis都是挺好玩的，不过我还没有玩出个花样来。为什么读中学大学时候要抑制女生玩游戏呢，其实女生也是很喜欢玩游戏的，也挺喜欢编程的。
　　Python 中rabbitmq的主要模块有pika, celery, haigha
　　就是相当于把队列放在一台远程服务器上，然后连上这个远程服务器，取数据，发数据。

• 1.         基本版
  

　　#######生产者#########
　　import pika
　　#申明链接
　　connection  =  pika. BlockingConnection(pika.ConnectionParameters(“localhost”))
　　#申明管道
　　channel = connection.channel()
　　#申明queue
　　channel.queue_declare(queue=’hello’)
　　#发消息
　　channel.basic_publish(exchange = “”,
　　Routing_key = “hello”, #queue 的名字
　　Body = “hello word” #所发的消息内容
　　)
　　connection.close()
　　#######消费者#########
　　import pika
　　connection = pika.BlockingConnection(pika.ConnectionParameters(‘localhost’))
　　channel = connection.channel()
　　#因为不确定是消费者还是生产者谁先运行，所以需要声明。防止生产者端后启动报错
　　channel.queue_declare(queue=’hello’)
　　#定义回调函数，一旦消息来了，就先执行这个函数来处理消息
　　def callback(ch,method,properties,body): #ch是管道内寸对象地址
　　print(“have received the message”,body)
　　ch.basic_ack(delivery_tag = method.delivery_tag) # 需要手动确认，所以no_ack = False, 那么消费者端需要手动和生产者端确认。
　　channel.basic_consume(callback,
　　queue= “hello”,
　　no_ack = True #不管消息处理完没有，都不会给消费者一个回复，默认不加。no_ack=False, 这样的话服务器端不会将消息删除，知道消息被处理，收到消费者的确认之后，才会删除消息。
　　)
　　channel.start_consuming() #一启动就会一直收消息，要是没有消息，那就等待消息来
　　如果一个生产者对应多个消费者，那么消费者之间会进行负载均衡，消息发送轮询。
　　队列和消息持久化：

消费者死机	服务器死机
那生产者应该默认这条消息没有被处理完，所以不能把这条消息从消息队列中删除。 　　消费者： 　　basic_consume 中no_ack =False, 　　callback 函数中: ch.basic_ack(delivery_tag = method.delivery_tag)	在申明队列的时候, 声明队列是持久化，但是队列中的消息会没有。消费者端和生产者端都需要写。 　　队列持久化 　　生产者、消费者： 　　Channel.queue_decalre(queue=’hello’, durable=True) 　　消息持久化 　　生产者： 　　poperties = pika.BasicProperties( 　　dlivery_mode = 2)

• 2.         广播模式
  

　　公平消息分发：根据处理器的处理能力来分配消息的量
　　消费者：
　　Channel.basic_qos(perfetch_count =1) #消费者要是还有一条消息的时候，生产者就不给他发送
　　要实现一个生产者发送消息，所有消费者都收到，那么需要广播模式。Exchange的类型，决定了消息如何被处理。
　　fanout: 所有绑定在这个exchange的queue都可以接收消息

生产者

消费者

import pika 　　import sys 　　connection = pika.BlockingConnection(pika.ConnectionParameters 　　(host = ‘localhost’)) 　　channel = connection.channel() 　　channel.exchange_declare( 　　exchange = ‘logs’, 　　type = ‘fanout’) 　　message = ‘info:hello world’ 　　channel.basix_publish(exchange = ‘logs’, 　　routing_key = ‘’, 　　body = message) 　　print(“[x] sent %r”%message) 　　connection.close()

import pika 　　connection = pika.BlockingConnection(pika.ConnectionParameters(host = “localhost”)) 　　channel = connection.channel() 　　channel.exchange_declare(exchange=’logs’, 　　type = ‘fanout’) 　　result =  channel.queue_declare(exclusive = True) 　　queue_name = result.method.queue 　　#不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除 　　def callback(ch, method, properties, body): 　　print(“[x] %r” %body) 　　channel.basic_consume(callback, 　　queue=queue_name, 　　no_ack = True 　　) 　　channel.start_consuming()

• 3.         选择模式
  

　　接收者只接收自己需要的信息
　　direct:通过routinekey和exchange决定那个唯一的queue可以接收消息
　　channel.exchange_declare(exchange=”direct_logs”, type = “direct”)
　　severity = sys.argv[1] if len(sys.argv) > 1 else “info” #选择接收，有不同级别的信息，info的信息发到一个队列，error的信息发送到另外一个队列。

生产者

消费者

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()

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) 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()

• 4.         更细致的消息过滤模式
  

　　topic：所有含有routingkey和routingkey所bind的queue都可以收消息

生产者

消费者

import pika import sys connectionpika.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()

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()