1#!/usr/bin/env python 2import pika 3import json 4 5from callback import callback 6 7 8class RabbitQueue: 9 def __init__(self): 10 self.channel = None 11 12 def connect(self): 13 credit = pika.PlainCredentials(username='admin', password='admin') 14 self.channel = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.3.19', port=5672, credentials=credit)).channel() 15 16 @staticmethod 17 def callback(channel, method, properties, body): 18 """callback函数需要自定义：返回结果：body为消息队列获取结果""" 19 receive = body.decode() 20 print(channel.__dict__) 21 print(receive) 22 print(method) 23 print(properties) 24 channel.basic_ack(delivery_tag=method.delivery_tag) 25 26 def image_enqueue(self, queue_name, image_list): 27 """推送数据至Rabbitmq消息队列""" 28 self.connect() 29 channel = self.channel 30 channel.queue_declare(queue=queue_name, durable=True) # 声明RPC请求队列,durable=True队列持久化 31 channel.basic_publish( 32 exchange='', 33 routing_key=queue_name, 34 body=json.dumps(image_list, ensure_ascii=False), 35 properties=pika.BasicProperties( 36 delivery_mode=2, # 消息持久化 37 ) 38 ) 39 channel.close() 40 41 def bpop_queue(self, queue_name, timeout=0): 42 """从消息队列获取数据""" 43 self.connect() 44 channel = self.channel 45 channel.queue_declare(queue=queue_name, durable=True) 46 # 需要自定义callback函数 47 channel.basic_consume(on_message_callback=callback, queue=queue_name, auto_ack=False) 48 channel.start_consuming() 49 50 51if __name__ == '__main__': 52 obj = RabbitQueue() 53 obj.image_enqueue(queue_name='_device_image_', image_list='hello world') 54 55 obj.bpop_queue(queue_name='_device_image_') # 阻塞等待
简单模式：
1# #########################基于简单模式的 生产者 ######################### 2#!/usr/bin/env python 3import pika 4 5# 封装 socket通信实现 6connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.253.128',port=5672)) 7 8# 创建通道对象 9channel = connection.channel() 10 11# 创建一个队列：名字是hello 12channel.queue_declare(queue='hello') 13 14# 向队列hello里丢东西 15channel.basic_publish(exchange='', 16 routing_key='hello', 17 body='Hello World!') 18 19print(" [x] Sent 'Hello World!'") 20connection.close() 21 22# ##########################基于简单模式的 消费者 ########################## 23import pika 24 25 26connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.253.128',port=5672)) 27channel = connection.channel() 28 29channel.queue_declare(queue='hello') 30 31 32def callback(ch, method, properties, body): 33 print(" [x] Received %r" % body) 34 35# 如果能从hello这个队列获取到数据就执行callback，否则继续往下走 36channel.basic_consume(callback, 37 queue='hello', 38 no_ack=True)no_ack参数：如果为False，当消费者服务器挂掉了，那么rabbitmq会重新将该任务添加到任务队列中 39 40print(' [*] Waiting for messages. To exit press CTRL+C') 41channel.start_consuming()
此时服务端的代码可以这么写：
1import pika 2 3connection = pika.BlockingConnection(pika.ConnectionParameters( 4 host='10.211.55.4')) 5channel = connection.channel() 6 7channel.queue_declare(queue='hello') 8 9def callback(ch, method, properties, body): 10 print(" [x] Received %r" % body) 11 import time 12 time.sleep(10) 13 print 'ok' 14 ch.basic_ack(delivery_tag = method.delivery_tag) 15 16channel.basic_consume(callback, 17 queue='hello', 18 no_ack=False) 19 20print(' [*] Waiting for messages. To exit press CTRL+C') 21channel.start_consuming() 22 23durable模式：信息不丢失 24# 生产者 25#!/usr/bin/env python 26import pika 27 28connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4')) 29channel = connection.channel() 30 31# make message persistent 32channel.queue_declare(queue='hello', durable=True) 33 34channel.basic_publish(exchange='', 35 routing_key='hello', 36 body='Hello World!', 37 properties=pika.BasicProperties( 38 delivery_mode=2, # make message persistent 39 )) 40print(" [x] Sent 'Hello World!'") 41connection.close() 42 43 44# 消费者 45#!/usr/bin/env python 46# -*- coding:utf-8 -*- 47import pika 48 49connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4')) 50channel = connection.channel() 51 52# make message persistent 53channel.queue_declare(queue='hello', durable=True) 54 55 56def callback(ch, method, properties, body): 57 print(" [x] Received %r" % body) 58 import time 59 time.sleep(10) 60 print 'ok' 61 ch.basic_ack(delivery_tag = method.delivery_tag) 62 63channel.basic_consume(callback, 64 queue='hello', 65 no_ack=False) 66 67print(' [*] Waiting for messages. To exit press CTRL+C') 68channel.start_consuming()
消息获取顺序
默认消息队列里的数据是按照顺序被消费者拿走，例如：消费者1 去队列中获取 奇数 序列的任务，消费者2去队列中获取 偶数 序列的任务。
channel.basic_qos(prefetch_count=1) 表示谁来谁取，不再按照奇偶数排列
enchange模型
一、发布订阅模式（fanout）
发布订阅和简单的消息队列区别在于，发布订阅会将消息发送给所有的订阅者，而消息队列中的数据被消费一次便消失。所以，RabbitMQ实现发布和订阅时，会为每一个订阅者创建一个队列，而发布者发布消息时，会将消息放置在所有相关队列中。
1# 生产者 2#!/usr/bin/env python 3import pika 4import sys 5 6connection = pika.BlockingConnection(pika.ConnectionParameters( 7 host='localhost')) 8channel = connection.channel() 9 10channel.exchange_declare(exchange='logs', 11 exchange_type='fanout') 12 13message = ' '.join(sys.argv[1:]) or "info: Hello World!" 14channel.basic_publish(exchange='logs', 15 routing_key='', 16 body=message) 17print(" [x] Sent %r" % message) 18connection.close() 19 20 21# 消费者 22#!/usr/bin/env python 23import pika 24 25connection = pika.BlockingConnection(pika.ConnectionParameters( 26 host='localhost')) 27channel = connection.channel() 28 29channel.exchange_declare(exchange='logs', 30 exchange_type='fanout') 31 32result = channel.queue_declare(exclusive=True) 33queue_name = result.method.queue 34 35channel.queue_bind(exchange='logs', 36 queue=queue_name) 37 38print(' [*] Waiting for logs. To exit press CTRL+C') 39 40def callback(ch, method, properties, body): 41 print(" [x] %r" % body) 42 43channel.basic_consume(callback, 44 queue=queue_name, 45 no_ack=True) 46 47channel.start_consuming()
二、关键字模式（direct）
之前事例，发送消息时明确指定某个队列并向其中发送消息，RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字 判定应该将数据发送至指定队列。
1#!/usr/bin/env python 2import pika 3import sys 4 5connection = pika.BlockingConnection(pika.ConnectionParameters( 6 host='localhost')) 7channel = connection.channel() 8 9channel.exchange_declare(exchange='direct_logs', 10 exchange_type='direct') 11 12result = channel.queue_declare(exclusive=True) 13queue_name = result.method.queue 14 15severities = sys.argv[1:] 16if not severities: 17 sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0]) 18 sys.exit(1) 19 20for severity in severities: 21 channel.queue_bind(exchange='direct_logs', 22 queue=queue_name, 23 routing_key=severity) 24 25print(' [*] Waiting for logs. To exit press CTRL+C') 26 27def callback(ch, method, properties, body): 28 print(" [x] %r:%r" % (method.routing_key, body)) 29 30channel.basic_consume(callback, 31 queue=queue_name, 32 no_ack=True) 33 34channel.start_consuming()
三、模糊匹配（topic）
在topic类型下，可以让队列绑定几个模糊的关键字，之后发送者将数据发送到exchange，exchange将传入”路由值“和 ”关键字“进行匹配，匹配成功，则将数据发送到指定队列。
1#!/usr/bin/env python 2import pika 3import sys 4 5connection = pika.BlockingConnection(pika.ConnectionParameters( 6 host='localhost')) 7channel = connection.channel() 8 9channel.exchange_declare(exchange='topic_logs', 10 exchange_type='topic') 11 12result = channel.queue_declare(exclusive=True) 13queue_name = result.method.queue 14 15binding_keys = sys.argv[1:] 16if not binding_keys: 17 sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0]) 18 sys.exit(1) 19 20for binding_key in binding_keys: 21 channel.queue_bind(exchange='topic_logs', 22 queue=queue_name, 23 routing_key=binding_key) 24 25print(' [*] Waiting for logs. To exit press CTRL+C') 26 27def callback(ch, method, properties, body): 28 print(" [x] %r:%r" % (method.routing_key, body)) 29 30channel.basic_consume(callback, 31 queue=queue_name, 32 no_ack=True) 33 34channel.start_consuming()
基于rabbitmq的RPC
1#!/usr/bin/env python 服务端 2import pika 3 4# 建立连接，服务器地址为localhost，可指定ip地址 5connection = pika.BlockingConnection(pika.ConnectionParameters( 6 host='192.168.253.128', port=5672)) 7 8# 建立会话 9channel = connection.channel() 10 11# 声明RPC请求队列 12channel.queue_declare(queue='rpc_queue') 13 14 15# 数据处理方法 16def fib(n): 17 if n == 0: 18 return 0 19 elif n == 1: 20 return 1 21 else: 22 return fib(n - 1) + fib(n - 2) 23 24 25# 对RPC请求队列中的请求进行处理 26def on_request(ch, method, props, body): 27 n = int(body) 28 29 print(" [.] fib(%s)" % n) 30 31 # 调用数据处理方法 32 response = fib(n) 33 34 # 将处理结果(响应)发送到回调队列 35 ch.basic_publish(exchange='', 36 routing_key=props.reply_to, 37 properties=pika.BasicProperties(correlation_id= \ 38 props.correlation_id), 39 body=str(response)) 40 ch.basic_ack(delivery_tag=method.delivery_tag) 41 42 43# 负载均衡，同一时刻发送给该服务器的请求不超过一个 44channel.basic_qos(prefetch_count=1) 45 46channel.basic_consume(on_request, queue='rpc_queue') 47 48print(" [x] Awaiting RPC requests") 49channel.start_consuming() 50 51#!/usr/bin/env python 52import pika 53import uuid 54 55class FibonacciRpcClient(object): 56 def __init__(self): 57 # 建立连接，指定服务器的ip地址 58 self.connection = pika.BlockingConnection(pika.ConnectionParameters( 59 host='192.168.253.128', port=5672)) 60 61 # 建立一个会话，每个channel代表一个会话任务 62 self.channel = self.connection.channel() 63 64 # 声明回调队列，再次声明的原因是，服务器和客户端可能先后开启，该声明是幂等的，多次声明，但只生效一次 65 result = self.channel.queue_declare(exclusive=True) 66 # 将次队列指定为当前客户端的回调队列 67 self.callback_queue = result.method.queue 68 69 # 客户端订阅回调队列，当回调队列中有响应时，调用`on_response`方法对响应进行处理; 70 self.channel.basic_consume(self.on_response, no_ack=True, 71 queue=self.callback_queue) 72 73 # 对回调队列中的响应进行处理的函数 74 def on_response(self, ch, method, props, body): 75 if self.corr_id == props.correlation_id: 76 self.response = body 77 78 # 发出RPC请求 79 def call(self, n): 80 81 # 初始化 response 82 self.response = None 83 84 # 生成correlation_id 85 self.corr_id = str(uuid.uuid4()) 86 87 # 发送RPC请求内容到RPC请求队列`rpc_queue`，同时发送的还有`reply_to`和`correlation_id` 88 self.channel.basic_publish(exchange='', 89 routing_key='rpc_queue', 90 properties=pika.BasicProperties( 91 reply_to=self.callback_queue, 92 correlation_id=self.corr_id, 93 ), 94 body=str(n)) 95 96 while self.response is None: 97 self.connection.process_data_events() 98 return int(self.response) 99 100# 建立客户端 101fibonacci_rpc = FibonacciRpcClient() 102 103# 发送RPC请求 104print("开始发送数据") 105response = fibonacci_rpc.call(30) 106print(" [.] Got %r" % response)