python学习-day11
Posted
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了python学习-day11相关的知识,希望对你有一定的参考价值。
一、Rabbitmq
RabbitMQ 即一个消息队列,主要是用来实现应用程序的异步和解耦,同时也能起到消息缓冲,消息分发的作用。RabbitMQ使用的是AMQP协议,它是一种二进制协议。默认启动端口 5672。在 RabbitMQ 中,如下图结构:
- 左侧 P 代表 生产者,也就是往 RabbitMQ 发消息的程序。生产者需要完成的任务:
-
1 创建RabbitMQ连接 2 获取信道 3 声明交换器 4 创建消息 5 发布消息 6 关闭信道 7 关闭RabbitMQ连接
- 中间即是 RabbitMQ,其中包括了 交换机 和 队列。
- 右侧 C 代表 消费者,也就是往 RabbitMQ 拿消息的程序。消费者需要完成的任务:
1 创建RabbitMQ连接 2 获取信道 3 声明交换器 4 声明队列 5 队列和交换器绑定 6 消费信息 7 关闭信道 8 关闭RabbitMQ连接
- Exchange: 接受客户端发送的消息,并根据Binding将消息路由给服务器中的队列,Exchange分为direct, fanout, topic三种。
- Binding: 连接Exchange和Queue,包含路由规则。
- Queue: 消息队列,存储还未被消费的消息。
- Message: Header+Body
- Channel: 通道,执行AMQP的命令;一个连接可创建多个通道以节省资源
1. dircted exchange
路由键exchange,该交换机收到消息后会把消息发送到指定routing-key的queue中。rabbitmq内部默认有一个特殊的dircted exchange,该交换机的name是空字符串,所有queue都默认binding 到该交换机上。所有binding到该交换机上的queue,routing-key都和queue的name一样。
生产者:
1 import pika 2 credentials = pika.PlainCredentials(‘admin‘, ‘123456‘) 3 connection = pika.BlockingConnection(pika.ConnectionParameters( 4 ‘192.168.170.134‘,5672,‘/‘,credentials)) 5 channel = connection.channel() 6 7 # 声明queue 8 channel.queue_declare(queue=‘hello‘) 9 10 # n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange. 11 channel.basic_publish(exchange=‘‘, 12 routing_key=‘hello‘, 13 body=‘Hello World!‘) 14 print(" [x] Sent ‘Hello World!‘") 15 connection.close()
消费者:
1 import pika 2 credentials = pika.PlainCredentials(‘admin‘, ‘123456‘) 3 connection = pika.BlockingConnection(pika.ConnectionParameters( 4 ‘192.168.170.134‘,5672,‘/‘,credentials)) 5 channel = connection.channel() 6 def callback(ch, method, properties, body): 7 print(" [x] Received %r" % body) 8 9 10 channel.basic_consume(callback, 11 queue=‘hello‘, 12 no_ack=True) 13 14 print(‘ [*] Waiting for messages. To exit press CTRL+C‘) 15 channel.start_consuming()
队列绑定关键字,发送者将数据关键字发送到消息Exchange,Exchange根据关键字判定应该将数据发送至指定队列。
生产者:
1 import pika,sys 2 3 credentials = pika.PlainCredentials(‘admin‘, ‘123456‘) 4 connection = pika.BlockingConnection(pika.ConnectionParameters( 5 ‘192.168.170.134‘,5672,‘/‘,credentials)) 6 channel = connection.channel() 7 channel.exchange_declare(exchange=‘direct_logs‘, 8 type=‘direct‘) 9 10 severity = sys.argv[1] if len(sys.argv) > 1 else ‘info‘ 11 message = ‘ ‘.join(sys.argv[2:]) or ‘Hello World!‘ 12 channel.basic_publish(exchange=‘direct_logs‘, 13 routing_key=severity, 14 body=message) 15 print(" [x] Sent %r:%r" % (severity, message)) 16 connection.close()
消费者:
1 import pika,sys 2 3 credentials = pika.PlainCredentials(‘admin‘, ‘123456‘) 4 connection = pika.BlockingConnection(pika.ConnectionParameters( 5 ‘192.168.170.134‘,5672,‘/‘,credentials)) 6 channel = connection.channel() 7 8 channel.exchange_declare(exchange=‘direct_logs‘, 9 type=‘direct‘) 10 11 result = channel.queue_declare(exclusive=True) 12 queue_name = result.method.queue 13 14 severities = sys.argv[1:] 15 if not severities: 16 sys.stderr.write("Usage: %s [info] [warning] [error]\\n" % sys.argv[0]) 17 sys.exit(1) 18 19 for severity in severities: 20 channel.queue_bind(exchange=‘direct_logs‘, 21 queue=queue_name, 22 routing_key=severity) 23 24 print(‘ [*] Waiting for logs. To exit press CTRL+C‘) 25 26 27 def callback(ch, method, properties, body): 28 print(" [x] %r:%r" % (method.routing_key, body)) 29 30 31 channel.basic_consume(callback, 32 queue=queue_name, 33 no_ack=True) 34 35 channel.start_consuming()
运行结果:
2. fanout exchange
发布/订阅exchange ,发布订阅和简单的消息队列区别在于,发布订阅会将消息发送给所有的订阅者,而消息队列中的数据被消费一次便消失。所以,RabbitMQ实现发布和订阅时,会为每一个订阅者创建一个队列,二发布者发布消息时,会将消息放置在所有相关队列中。任何发送到Fanout Exchange的消息都会被转发到与该Exchange绑定(Binding)的所有Queue上。
生产者:
1 import pika,sys 2 3 credentials = pika.PlainCredentials(‘admin‘, ‘123456‘) 4 connection = pika.BlockingConnection(pika.ConnectionParameters( 5 ‘192.168.170.134‘,5672,‘/‘,credentials)) 6 channel = connection.channel() 7 # 声明queue 8 channel.exchange_declare(exchange=‘logs‘, 9 type=‘fanout‘) 10 11 message = ‘ ‘.join(sys.argv[1:]) or "info: Hello World!" 12 channel.basic_publish(exchange=‘logs‘, 13 routing_key=‘‘, 14 body=message) 15 print(" [x] Sent %r" % message) 16 connection.close()
消费者:
1 import pika 2 3 credentials = pika.PlainCredentials(‘admin‘, ‘123456‘) 4 connection = pika.BlockingConnection(pika.ConnectionParameters( 5 ‘192.168.170.134‘,5672,‘/‘,credentials)) 6 channel = connection.channel() 7 8 result = channel.queue_declare(exclusive=True) # 不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除 9 queue_name = result.method.queue 10 11 channel.queue_bind(exchange=‘logs‘, 12 queue=queue_name) 13 14 print(‘ [*] Waiting for logs. To exit press CTRL+C‘) 15 16 17 def callback(ch, method, properties, body): 18 print(" [x] %r" % body) 19 20 21 channel.basic_consume(callback, 22 queue=queue_name, 23 no_ack=True) 24 25 channel.start_consuming()
3. topic exchange
在topic类型下,可以让队列绑定几个模糊的关键字,之后发送者将数据发送到exchange,exchange将传入"路由值"和"关键字"进行匹配,匹配成功,则将数据发送到指定队列。
-
# :表示可以匹配0个或多个单词;
-
* :表示只能匹配一个单词。
生产者:
1 import pika,sys 2 3 credentials = pika.PlainCredentials(‘admin‘, ‘123456‘) 4 connection = pika.BlockingConnection(pika.ConnectionParameters( 5 ‘192.168.170.134‘,5672,‘/‘,credentials)) 6 channel = connection.channel() 7 8 channel.exchange_declare(exchange=‘topic_logs‘, 9 type=‘topic‘) 10 11 routing_key = sys.argv[1] if len(sys.argv) > 1 else ‘anonymous.info‘ 12 message = ‘ ‘.join(sys.argv[2:]) or ‘Hello World!‘ 13 channel.basic_publish(exchange=‘topic_logs‘, 14 routing_key=routing_key, 15 body=message) 16 print(" [x] Sent %r:%r" % (routing_key, message)) 17 connection.close()
消费者:
1 import pika,sys 2 3 credentials = pika.PlainCredentials(‘admin‘, ‘123456‘) 4 connection = pika.BlockingConnection(pika.ConnectionParameters( 5 ‘192.168.170.134‘,5672,‘/‘,credentials)) 6 channel = connection.channel() 7 8 channel.exchange_declare(exchange=‘topic_logs‘, 9 type=‘topic‘) 10 11 result = channel.queue_declare(exclusive=True) 12 queue_name = result.method.queue 13 14 binding_keys = sys.argv[1:] 15 if not binding_keys: 16 sys.stderr.write("Usage: %s [binding_key]...\\n" % sys.argv[0]) 17 sys.exit(1) 18 19 for binding_key in binding_keys: 20 channel.queue_bind(exchange=‘topic_logs‘, 21 queue=queue_name, 22 routing_key=binding_key) 23 24 print(‘ [*] Waiting for logs. To exit press CTRL+C‘) 25 26 27 def callback(ch, method, properties, body): 28 print(" [x] %r:%r" % (method.routing_key, body)) 29 30 31 channel.basic_consume(callback, 32 queue=queue_name, 33 no_ack=True) 34 35 channel.start_consuming()
二、基于rabbitmq的RPC
基于rabbitmq的rpc实现流程:
(1)首先客户端发送一个reply_to和corrention_id的请求,发布到RPC队列中;
(2)服务器端处理这个请求,并把处理结果发布到一个回调Queue,此Queue的名称应当与reply_to的名称一致
(3)客户端从回调Queue中得到先前correlation_id设定的值的处理结果。如果碰到和先前不一样的corrention_id的值,将会忽略而不是抛出异常。
对于上面所提到的回调Queue中的消费处理使用的是BasicProperties类。
服务端:
1 import pika 2 3 cre_publiser = pika.PlainCredentials(‘admin‘, ‘123456‘) 4 conn_para = pika.ConnectionParameters(‘192.168.170.134‘,5672,‘/‘,cre_publiser) 5 connection = pika.BlockingConnection(conn_para) 6 7 # 建立会话 8 channel = connection.channel() 9 10 # 声明RPC请求队列 11 channel.queue_declare(queue=‘rpc_queue‘) 12 13 # 数据处理方法 14 def fib(n): 15 if n == 0: 16 return 0 17 elif n == 1: 18 return 1 19 else: 20 return fib(n-1) + fib(n-2) 21 22 # 对RPC请求队列中的请求进行处理 23 def on_request(ch, method, props, body): 24 n = int(body) 25 26 print(" [.] fib(%s)" % n) 27 28 # 调用数据处理方法 29 response = fib(n) 30 31 # 将处理结果(响应)发送到回调队列 32 ch.basic_publish(exchange=‘‘, 33 routing_key=props.reply_to, 34 properties=pika.BasicProperties(correlation_id=props.correlation_id), 35 body=str(response)) 36 ch.basic_ack(delivery_tag = method.delivery_tag) 37 38 # 负载均衡,同一时刻发送给该服务器的请求不超过一个 39 channel.basic_qos(prefetch_count=1) 40 41 channel.basic_consume(on_request, 42 queue=‘rpc_queue‘) 43 44 print(" [x] Awaiting RPC requests") 45 channel.start_consuming()
客户端:
1 import pika 2 import uuid 3 class FibonacciRpcClient(object): 4 def __init__(self): 5 self.cre_publiser = pika.PlainCredentials(‘admin‘, ‘123456‘) 6 self.conn_para = pika.ConnectionParameters(‘192.168.170.134‘,5672,‘/‘,self.cre_publiser) 7 self.connection = pika.BlockingConnection(self.conn_para) 8 9 self.channel = self.connection.channel() 10 11 result = self.channel.queue_declare(exclusive=True) 12 self.callback_queue = result.method.queue 13 14 self.channel.basic_consume(self.on_response, 15 no_ack=True, 16 queue=self.callback_queue) 17 18 def on_response(self, ch, method, props, body): 19 if self.corr_id == props.correlation_id: 20 self.response = body 21 22 def call(self, n): 23 self.response = None 24 self.corr_id = str(uuid.uuid4()) 25 self.channel.basic_publish(exchange=‘‘, 26 routing_key=‘rpc_queue‘, 27 properties=pika.BasicProperties( 28 reply_to=self.callback_queue, 29 correlation_id=self.corr_id, 30 ), 31 body=str(n)) 32 while self.response is None: 33 self.connection.process_data_events() 34 return int(self.response) 35 36 37 fibonacci_rpc = FibonacciRpcClient() 38 39 print(" [x] Requesting fib(6)") 40 response = fibonacci_rpc.call(6) 41 print(" [.] Got %r" % response)
以上是关于python学习-day11的主要内容,如果未能解决你的问题,请参考以下文章