关于rabbitmq:Python对RabbitMQ的简单使用

(一) RabbitMQ的简介

RabbitMq 是实现了高级音讯队列协定（AMQP）(Advanced Message Queuing Protocol)的开源音讯代理中间件。音讯队列是一种应用程序对应用程序的通行形式，应用程序通过写音讯，将消息传递于队列，由另一应用程序读取实现通信。而作为中间件的 RabbitMq 无疑是目前最风行的音讯队列之一。目前应用较多的音讯队列有ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ。

RabbitMQ总体架构

PS：生产者和消费者可能在不同的程序或主机中，当然也有可能一个程序有可能既是生产者，也是消费者。

RabbitMq 利用场景宽泛：

零碎的高可用：日常生活当中各种商城秒杀，高流量，高并发的场景。当服务器接管到如此大量申请解决业务时，有宕机的危险。某些业务可能极其简单，但这部分不是高时效性，不须要立刻反馈给用户，咱们能够将这部分解决申请抛给队列，让程序后置去解决，加重服务器在高并发场景下的压力。
分布式系统，集成系统，子系统之间的对接，以及架构设计中经常须要思考音讯队列的利用。

(二) RabbitMQ的装置

这里通过官网下载须要的版本：RabbitMQ官网网址
鉴于官网拜访下载比较慢，贴一个云盘地址：百度云盘地址

Mac的话，能够间接应用brew装置,brew相似ubuntu下的apt-get性能

# brew install rabbitmq# 启动rabbitmq： rabbitmq-server start# 进行rabbitmq： rabbitmq-server stop# 重启rabbitmq： rabbitmq-server restart

启动后，就能够登陆后盾治理页面了，浏览器输出ip:15672

自带的明码和用户名都是guest

(三) python操作RabbitMQ

python中应用pika操作RabbitMQ

pip3 install pika

(四) RabbitMQ简略模式

上代码

# coding=utf-8### 生产者import pikaimport timeuser_info = pika.PlainCredentials('guest', 'guest')#用户名和明码connection = pika.BlockingConnection(pika.ConnectionParameters('localhost', 5672, '/', user_info))#连贯服务器上的RabbitMQ服务# 创立一个channelchannel = connection.channel()# 如果指定的queue不存在，则会创立一个queue，如果曾经存在 则不会做其余动作，官网举荐，每次应用时都能够加上这句channel.queue_declare(queue='hello')for i in range(0, 100):    channel.basic_publish(exchange='',#以后是一个简略模式，所以这里设置为空字符串就能够了                          routing_key='hello',# 指定音讯要发送到哪个queue                          body='{}'.format(i)# 指定要发送的音讯                          )    time.sleep(1)    # 敞开连贯# connection.close()

PS：RabbitMQ中所有的音讯都要先通过交换机，空字符串示意应用默认的交换机

# coding=utf-8### 消费者import pikauser_info = pika.PlainCredentials('guest', 'guest')connection = pika.BlockingConnection(pika.ConnectionParameters('localhost', 5672, '/', user_info))channel = connection.channel()# 如果指定的queue不存在，则会创立一个queue，如果曾经存在 则不会做其余动作，生产者和消费者都做这一步的益处是# 这样生产者和消费者就没有必要的先后启动程序了channel.queue_declare(queue='hello')# 回调函数def callback(ch, method, properties, body):    print('消费者收到:{}'.format(body))# channel: 蕴含channel的所有属性和办法# method: 蕴含 consumer_tag, delivery_tag, exchange, redelivered, routing_key# properties: basic_publish 通过 properties 传入的参数# body: basic_publish发送的音讯channel.basic_consume(queue='hello',  # 接管指定queue的音讯                      auto_ack=True,  # 指定为True，示意音讯接管到后主动给音讯发送方回复确认，已收到音讯                      on_message_callback=callback  # 设置收到音讯的回调函数                      )print('Waiting for messages. To exit press CTRL+C')# 始终处于期待接管音讯的状态，如果没收到音讯就始终处于阻塞状态，收到音讯就调用下面的回调函数channel.start_consuming()

对于下面的这种模式，有一下两个不好的中央：

一个是在咱们的消费者还没开始生产完队列里的音讯，如果这时rabbitmq服务挂了，那么音讯队列里的音讯将会全副失落，解决办法是在申明队列时，申明队列为可长久化存储队列，并且在生产者在将音讯插入到音讯队列时，设置音讯长久化存储，具体如下

# coding=utf-8### 生产者import pikaimport timeuser_info = pika.PlainCredentials('guest', 'guest')connection = pika.BlockingConnection(pika.ConnectionParameters('localhost', 5672, '/', user_info))# 创立一个channelchannel = connection.channel()# 如果指定的queue不存在，则会创立一个queue，如果曾经存在 则不会做其余动作，官网举荐，每次应用时都能够加上这句channel.queue_declare(queue='durable_queue',durable=True)#PS：这里不同种队列不容许名字雷同for i in range(0, 100):    channel.basic_publish(exchange='',                          routing_key='durable_queue',                          body='{}'.format(i),                          properties=pika.BasicProperties(delivery_mode=2)                          )# 敞开连贯# connection.close()

消费者与下面的消费者没有什么不同，具体的就是生产申明的队列，也要是可长久化的队列，还有就是，即便在生产者插入音讯时，设置以后音讯长久化存储（properties=pika.BasicProperties(delivery_mode=2)），并不能百分百保障音讯真的被长久化，因为RabbitMQ挂掉的时候它可能还保留在缓存中，没来得及同步到磁盘中

在生产者插入音讯后，立即进行rabbitmq，并重新启动，其实咱们在web治理页面也可看到未被生产的信息，当然在启动消费者后也胜利接管到了音讯

下面说的第二点不好就是，如果在消费者获取到队列里的音讯后，在回调函数的处理过程中，消费者忽然出错或程序解体等异样，那么就会造成这条音讯并未被理论失常的解决掉。为了解决这个问题，咱们只需在消费者basic_consume(auto_ack=False),并在回调函数中设置手动应答即可ch.basic_ack(delivery_tag=method.delivery_tag)，具体如下

# coding=utf-8### 消费者import pikaimport timeuser_info = pika.PlainCredentials('root', 'root')connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))channel = connection.channel()# 如果指定的queue不存在，则会创立一个queue，如果曾经存在 则不会做其余动作，生产者和消费者都做这一步的益处是# 这样生产者和消费者就没有必要的先后启动程序了channel.queue_declare(queue='queue')# 回调函数def callback(ch, method, properties, body):    time.sleep(5)    ch.basic_ack(delivery_tag=method.delivery_tag)    print('消费者收到:{}'.format(body.decode('utf-8')))# channel: 蕴含channel的所有属性和办法# method: 蕴含 consumer_tag, delivery_tag, exchange, redelivered, routing_key# properties: basic_publish 通过 properties 传入的参数# body: basic_publish发送的音讯channel.basic_consume(queue='queue',  # 接管指定queue的音讯                      auto_ack=False,  # 指定为False，示意勾销自动应答，交由回调函数手动应答                      on_message_callback=callback  # 设置收到音讯的回调函数                      )# 应答的实质是通知音讯队列能够将这条音讯销毁了print('Waiting for messages. To exit press CTRL+C')# 始终处于期待接管音讯的状态，如果没收到音讯就始终处于阻塞状态，收到音讯就调用下面的回调函数channel.start_consuming()

这里只须要配置消费者，生产者并不要批改

还有就是在上的应用形式在，都是一个生产者和一个消费者，还有一种状况就是，一个生产者和多个消费者，即多个消费者同时监听一个音讯队列，这时候队列里的音讯就是轮询散发（即如果音讯队列里有100条信息，如果有2个消费者，那么每个就会收到50条信息），然而在某些状况下，不同的消费者解决工作的能力是不同的，这时还依照轮询的形式散发音讯并不是很正当，那么只须要再配合手动应答的形式，设置消费者接管的音讯没有解决完，队列就不要给我放送新的音讯即可，具体配置形式如下：

# coding=utf-8### 消费者import pikaimport timeuser_info = pika.PlainCredentials('root', 'root')connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))channel = connection.channel()# 如果指定的queue不存在，则会创立一个queue，如果曾经存在 则不会做其余动作，生产者和消费者都做这一步的益处是# 这样生产者和消费者就没有必要的先后启动程序了channel.queue_declare(queue='queue')# 回调函数def callback(ch, method, properties, body):    time.sleep(0)#通过设置休眠工夫来模仿不同消费者的解决工夫    ch.basic_ack(delivery_tag=method.delivery_tag)    print('消费者收到:{}'.format(body.decode('utf-8')))# prefetch_count示意接管的音讯数量，当我接管的音讯没有解决完（用basic_ack标记音讯已处理完毕）之前不会再接管新的音讯了channel.basic_qos(prefetch_count=1)  # 还有就是这个设置必须在basic_consume之上，否则不失效channel.basic_consume(queue='queue',  # 接管指定queue的音讯                      auto_ack=False,  # 指定为False，示意勾销自动应答，交由回调函数手动应答                      on_message_callback=callback  # 设置收到音讯的回调函数                      )# 应答的实质是通知音讯队列能够将这条音讯销毁了print('Waiting for messages. To exit press CTRL+C')# 始终处于期待接管音讯的状态，如果没收到音讯就始终处于阻塞状态，收到音讯就调用下面的回调函数channel.start_consuming()

PS:这种状况必须敞开自动应答ack，改成手动应答。应用basicQos(perfetch=1)限度每次只发送不超过1条音讯到同一个消费者，消费者必须手动反馈告知队列，才会发送下一个

(五) RabbitMQ公布订阅模式

公布订阅会将音讯发送给所有的订阅者，而音讯队列中的数据被生产一次便隐没。所以，RabbitMQ实现公布和订阅时，会为每一个订阅者创立一个队列，而发布者公布音讯时，会将音讯搁置在所有相干队列中

这个模式中会引入交换机的概念，其实在RabbitMQ中，所有的生产者都不会间接把音讯发送到队列中，甚至生产者都不晓得音讯在收回后有没有发送到queue中，事实上，生产者只能将音讯发送给交换机，由交换机来决定发送到哪个队列中。

交换机的一端用来从生产者中接管音讯，另一端用来发送音讯到队列，交换机的类型规定了怎么解决接管到的音讯，公布订阅模式应用到的交换机类型为 fanout ，这种交换机类型非常简单，就是将接管到的音讯播送给已知的（即绑定到此交换机的）所有消费者。

当然，如果不想应用特定的交换机，能够应用 exchange=’’ 示意应用默认的交换机，默认的替换机会将音讯发送到 routing_key 指定的queue，能够参考简略模式。

上代码：

#生产者import pikauser_info = pika.PlainCredentials('root', 'root')connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))channel = connection.channel()# 创立一个指定名称的交换机，并指定类型为fanout，用于将接管到的音讯播送到所有queue中channel.exchange_declare(exchange='交换机', exchange_type='fanout')# 将音讯发送给指定的交换机，在fanout类型中，routing_key=''示意不必发送到指定queue中，# 而是将发送到绑定到此交换机的所有queuechannel.basic_publish(exchange='交换机', routing_key='', body='这是一条测试音讯')

#消费者import pikauser_info = pika.PlainCredentials('root', 'root')connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))channel = connection.channel()channel.exchange_declare(exchange='交换机', exchange_type='fanout')# 应用RabbitMQ给本人生成一个专有的queueresult = channel.queue_declare(queue='333')# result = channel.queue_declare(queue='', exclusive=True)queue_name = result.method.queue# 这里如果设置exclusive=True参数，那么该队列就是一个只有队列，在消费者完结后，该专有队列也会主动革除，如果queue=''没有设置名字的话，那么就会主动生成一个# 不会反复的队列名# 将queue绑定到指定交换机channel.queue_bind(exchange='交换机', queue=queue_name)print(' [*] Waiting for  message.')def callback(ch, method, properties, body):    print("消费者收到:{}".format(body.decode('utf-8')))channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)channel.start_consuming()

该模式与简略模式的还有一个区别就是，这里的音讯队列都是由消费者申明的，所以如果是生产者先启动，并将音讯发给交换机的画，这里的音讯就会失落，所以咱们也能够在消费者端申明队列并绑定交换机（不能是专有队列），所以认真想想，其实这所谓的公布订阅模式并没有说什么了不起，它不过是让交换机同时推送多条音讯给绑定的队列，咱们当然也能够在简略模式的根底上多进行几次basic_publish发送音讯到指定的队列。当然咱们这样做的话，可能就没方法做到由交换机的同时发送了，效率可能也没有一次basic_publish的高

(六) RabbitMQ RPC模式

上面实现由rpc近程调用加减运算

客户端

import pikaimport uuidimport jsonclass RPC(object):    def __init__(self):        self.call_id = None        self.response = None        user_info = pika.PlainCredentials('root', 'root')        self.connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))        self.channel = self.connection.channel()        # 创立一个此客户端专用的queue，用于接管服务端发过来的音讯        result = self.channel.queue_declare(queue='', exclusive=True)        self.callback_queue = result.method.queue        self.channel.basic_consume(            queue=self.callback_queue,            on_message_callback=self.on_response,            auto_ack=True)    def on_response(self, ch, method, props, body):        # 判断接管到的response是否属于对应request        if self.call_id == props.correlation_id:            self.response = json.loads(body.decode('utf-8')).get('result')    def call(self, func, param):        self.response = None        self.call_id = str(uuid.uuid4())  # 为该音讯指定uuid，相似于申请id        self.channel.queue_declare(queue='rpc_queue')        self.channel.basic_publish(            exchange='',            routing_key='rpc_queue',  # 将音讯发送到该queue            properties=pika.BasicProperties(                reply_to=self.callback_queue,  # 从该queue中取音讯                correlation_id=self.call_id,  # 为此次音讯指定uuid            ),            body=json.dumps(                {                    'func': func,                    'param': {'a': param[0], 'b': param[1]}                }            )        )                self.connection.process_data_events(time_limit=3)# 与start_consuming()类似，能够设置超时参数        return self.responserpc = RPC()print("发送音讯到消费者,期待返回后果")response = rpc.call(func='del', param=(1, 2))print("收到来自消费者返回的后果:{}".format(response))

服务端

import pikaimport jsonuser_info = pika.PlainCredentials('root', 'root')connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))channel = connection.channel()# 指定接管音讯的queuechannel.queue_declare(queue='rpc_queue')def add_number(a, b):    return a + bdef del_num(a, b):    return a - bexecute_map = {    'add': add_number,    'del': del_num}def on_request(ch, method, props, body):    body = json.loads(body.decode('utf-8'))    func = body.get('func')    param = body.get('param')    result = execute_map.get(func)(param.get('a'), param.get('b'))    print('进行{}运算，并将后果返回个消费者'.format(func))    ch.basic_publish(exchange='',  # 应用默认交换机                     routing_key=props.reply_to,  # response发送到该queue                     properties=pika.BasicProperties(                         correlation_id=props.correlation_id),  # 应用correlation_id让此response与申请音讯对应起来                     body=json.dumps({'result': result}))    ch.basic_ack(delivery_tag=method.delivery_tag)channel.basic_qos(prefetch_count=1)# 从rpc_queue中取音讯，而后应用on_request进行解决channel.basic_consume(queue='rpc_queue', on_message_callback=on_request)print(" [x] Awaiting RPC requests")channel.start_consuming()

(六) RabbitMQ RPC模式

上面实现由rpc近程调用加减运算

客户端

import pikaimport uuidimport jsonclass RPC(object):    def __init__(self):        self.call_id = None        self.response = None        user_info = pika.PlainCredentials('root', 'root')        self.connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))        self.channel = self.connection.channel()        # 创立一个此客户端专用的queue，用于接管服务端发过来的音讯        result = self.channel.queue_declare(queue='', exclusive=True)        self.callback_queue = result.method.queue        self.channel.basic_consume(            queue=self.callback_queue,            on_message_callback=self.on_response,            auto_ack=True)    def on_response(self, ch, method, props, body):        # 判断接管到的response是否属于对应request        if self.call_id == props.correlation_id:            self.response = json.loads(body.decode('utf-8')).get('result')    def call(self, func, param):        self.response = None        self.call_id = str(uuid.uuid4())  # 为该音讯指定uuid，相似于申请id        self.channel.queue_declare(queue='rpc_queue')        self.channel.basic_publish(            exchange='',            routing_key='rpc_queue',  # 将音讯发送到该queue            properties=pika.BasicProperties(                reply_to=self.callback_queue,  # 从该queue中取音讯                correlation_id=self.call_id,  # 为此次音讯指定uuid            ),            body=json.dumps(                {                    'func': func,                    'param': {'a': param[0], 'b': param[1]}                }            )        )                self.connection.process_data_events(time_limit=3)# 与start_consuming()类似，能够设置超时参数        return self.responserpc = RPC()print("发送音讯到消费者,期待返回后果")response = rpc.call(func='del', param=(1, 2))print("收到来自消费者返回的后果:{}".format(response))

服务端

import pikaimport jsonuser_info = pika.PlainCredentials('root', 'root')connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))channel = connection.channel()# 指定接管音讯的queuechannel.queue_declare(queue='rpc_queue')def add_number(a, b):    return a + bdef del_num(a, b):    return a - bexecute_map = {    'add': add_number,    'del': del_num}def on_request(ch, method, props, body):    body = json.loads(body.decode('utf-8'))    func = body.get('func')    param = body.get('param')    result = execute_map.get(func)(param.get('a'), param.get('b'))    print('进行{}运算，并将后果返回个消费者'.format(func))    ch.basic_publish(exchange='',  # 应用默认交换机                     routing_key=props.reply_to,  # response发送到该queue                     properties=pika.BasicProperties(                         correlation_id=props.correlation_id),  # 应用correlation_id让此response与申请音讯对应起来                     body=json.dumps({'result': result}))    ch.basic_ack(delivery_tag=method.delivery_tag)channel.basic_qos(prefetch_count=1)# 从rpc_queue中取音讯，而后应用on_request进行解决channel.basic_consume(queue='rpc_queue', on_message_callback=on_request)print(" [x] Awaiting RPC requests")channel.start_consuming()

(七) 结语

对于rabbitmq的模式还有Routing模式和Topics模式等，这里就不复述了，其实pika对于RabbitMQ的应用还有很多细节和参数值得深究。这篇博客也就是简略的记录下我对pika操作raabbitmq过程和简略的了解

参考链接：

https://www.cnblogs.com/guyuy...
https://blog.csdn.net/wohu110...