标签：
一、Rabbitmq
RabbitMQ 即一个消息队列，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。RabbitMQ使用的是AMQP协议，它是一种二进制协议。默认启动端口 5672。在 RabbitMQ 中，如下图结构：
1 创建RabbitMQ连接 2 获取信道 3 声明交换器 4 创建消息 5 发布消息 6 关闭信道 7 关闭RabbitMQ连接
1 创建RabbitMQ连接 2 获取信道 3 声明交换器 4 声明队列 5 队列和交换器绑定 6 消费信息 7 关闭信道 8 关闭RabbitMQ连接
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)
标签：
原文地址：http://www.cnblogs.com/iclq/p/5967777.html
