python 使用pika对接rabbitMQ

本文主要是介绍python 使用pika对接rabbitMQ，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

1、简易阐述原理

• 原则上，消息，只能有交换机传到队列，就像我们家里面的交换机道理一样。
• 有多个设备连接到交换机，那么，这个交换机把消息发给那个设备呢，就是根据交换机的类型来定。类型有：direct\topic\headers\fanout
• fanout：这个就是，所有的设备都能收到消息，就是广播。
• 此处定义一个名称为'logs'的'fanout'类型的exchange
• channel.exchange_declare(exchange='logs', exchange_type='fanout')
• rabbitMQ详细原理阐述
• rabbitMQ与redis性能对比
• rabbitMQ原理介绍

2、个人使用rabbitmq见解

• 发布：实质就是每个数据上面附带一个route_key，将数据发送到某个交换机X，发布数据的时候只需要将数据发送到交换机即可，这里的交换机相当于（家里拉网，然后把网线接到路由器上进口上，这个路由器中转站有了数据，然后可以再用多根网线插到路由器的出口，获取到网络，这里的网线就相当于rabbitmq中的queue，route_key的作用是可以根据这个值决定数据走哪个queue）
• 消费：实质从交换机X上面取的route_key等于某个值放到某个队列，然后从某个队列进行消费，不断进行循环

3、发布订阅

• 以下是我借鉴网上一位大牛写的代码，从新把不完善的地方完善了一下，实现后端对接celery异步消费功能，目前程序一直在服务器中稳定运行，没有出现异常，如果程序还有啥问题，欢迎指正哈
• rabbitmq调试web页面为：http://ip:15672/#/queues

#! /usr/bin/env python2
# .-*- coding:utf-8 .-*-import pika
import json
import datetimefrom multiprocessing import Process
from pika.exceptions import ChannelClosed
from pika.exceptions import ConnectionClosed
from project.tasks import *
from my_logger import Loggerlogger_name = 'rabbitmq.log'
logger = Logger(logger_name)# rabbitmq 配置信息
MQ_CONFIG = {"host": "ip","port": 5672,"vhost": "/","user": "haha",      # rabbitmq中添加的用户名"passwd": "haha",    # rabbitmq中添加的用户名对应密码"exchange": "web",  # 需要从那个交换机上面取数据的name
}class RabbitMQServer(object):# _instance_lock = threading.Lock()def __init__(self):self.recv_queu = ""self.recv_rout_key = ""self.send_serverid = ""self.exchange = MQ_CONFIG.get("exchange")self.connection = Noneself.channel = Nonedef reconnect(self):try:if self.channel and not self.channel.is_closed:self.channel.close()if self.connection and not self.connection.is_closed:self.connection.close()# 创建一个身份验证凭证credentials = pika.PlainCredentials(username=MQ_CONFIG.get("user"),password=MQ_CONFIG.get("passwd"))# 创建一个参数连接对象，heartbeat=0可以设置成rebbit永久连接，不然等没有数据传输，则会中断连接；# 建议heartbeat的值设置为5-16可以满足一般的需求，设置的太小频繁访问容易造成网络拥堵parameters = pika.ConnectionParameters(host=MQ_CONFIG.get("host"),port=MQ_CONFIG.get("port"),virtual_host=MQ_CONFIG.get("vhost"),credentials=credentials,heartbeat=0,socket_timeout=5)# 在Pika的异步核心方法之上创建一个层一直阻塞到预期响应有结果self.connection = pika.BlockingConnection(parameters)# 相当于创建一个数据传输管道self.channel = self.connection.channel()# 如果交换器不存在，则创建一个交换器；如果交换器存在，则验证是否和预期一致# 使用durable=True声明queue是持久化的，这样即便Rabb崩溃了重启后queue仍然存在其中的message不会丢失self.channel.exchange_declare(exchange=self.exchange, durable=True)if isinstance(self, RabbitComsumer):# 根据需要声明队列，exclusive=True 使用结束后会自动删除队列result = self.channel.queue_declare(queue=self.recv_queu, durable=True)# 获取队列名queue_name = result.method.queue# 实质意义是从交换机上面取一个routing_key=self.recv_rout_key的数值，放到queue里面self.channel.queue_bind(exchange=self.exchange, queue=queue_name, routing_key=self.recv_rout_key)# 设置服务质量，公平调度，同一时间，每个队列只给分配一个任务prefetch_count=1self.channel.basic_qos(prefetch_count=1)# 消费queue里面的数据self.channel.basic_consume(self.consumer_callback, queue=queue_name, no_ack=False)except Exception as e:logger.error('Reconnect Exception: %s' % str(e))class RabbitComsumer(RabbitMQServer):def __init__(self):super(RabbitComsumer, self).__init__()def consumer_callback(self, ch, method, properties, body):""":param ch: 通道对象:param method: 可以获取队列数据的附带参数值，比如route_key:param properties::param body: 队列数据值:return:"""logger.info("rout_key: %s, body: %s, method.routing_key: %s" %(str(self.recv_rout_key), str(body), str(method.routing_key)))# 和deley相比apply_async可以控制控制任务执行的参数，异步分配任务if method.routing_key == "websusingle_urgent":get_subdomain.apply_async(args=[body])elif method.routing_key == "ICP_urgent":get_icp.apply_async(args=[body])elif method.routing_key == "whois_urgent":get_whois.apply_async(args=[body])elif method.routing_key == "dns_urgent":get_dns.apply_async(args=[body])elif method.routing_key == "passdns_urgent":get_passdns.apply_async(args=[body])else:self.channel.basic_publish(exchange='',  routing_key='info_webasset_error', body=body)# 保证消息不丢失，如果没有回复则重新添加任务ch.basic_ack(delivery_tag=method.delivery_tag)def start_consumer(self):while True:try:self.reconnect()self.channel.start_consuming()except ConnectionClosed as e:logger.error("ConnectionClosed Exception: %s" % str(e))self.reconnect()time.sleep(2)except ChannelClosed as e:logger.error("ChannelClosed Exception: %s" % str(e))self.reconnect()time.sleep(2)except Exception as e:logger.error("Other Exception: %s" % str(e))self.reconnect()time.sleep(2)@classmethoddef run(cls, info):consumer = cls()consumer.recv_queu = info[0]consumer.recv_rout_key = info[1]consumer.start_consumer()class RabbitPublisher(RabbitMQServer):""" 发布队列，本程序目前尚未使用 """def __init__(self):super(RabbitPublisher, self).__init__()def start_publish(self):self.reconnect()i = 1while True:message = {"value": i}message = dict_to_json(message)try:self.channel.basic_publish(exchange=self.exchange, routing_key=self.send_serverid, body=message)i += 1except ConnectionClosed as e:self.reconnect()time.sleep(2)except ChannelClosed as e:self.reconnect()time.sleep(2)except Exception as e:self.reconnect()time.sleep(2)@classmethoddef run(cls, send_serverid):publish = cls()publish.send_serverid = send_serveridpublish.start_publish()class CJsonEncoder(json.JSONEncoder):def default(self, obj):if isinstance(obj, datetime.datetime):return obj.strftime('%Y-%m-%d %H:%M:%S')elif isinstance(obj, datetime.date):return obj.strftime("%Y-%m-%d")else:return json.JSONEncoder.default(self, obj)def dict_to_json(po):jsonstr = json.dumps(po, ensure_ascii=False, cls=CJsonEncoder)return jsonstrdef json_to_dict(jsonstr):if isinstance(jsonstr, bytes):jsonstr = jsonstr.decode("utf-8")d = json.loads(jsonstr)return ddef work(func, items):for item in items:p = Process(target=func, args=(item,))p.start()if __name__ == '__main__':# 这里分别用两个线程去连接和发送# threading.Thread(target=RabbitComsumer.run, args=(recv_serverid,)).start()# threading.Thread(target=RabbitPublisher.run, args=(send_serverid,)).start()# 如果CPU是多核，资源够用的情况下建议使用多进程# list里面放两个元素，分别对应消费需要的queue和消费的route_keytemp1 = ["queue_name1", "route_key1"]temp2 = ["queue_name2", "route_key2"]items = [temp1, temp2]work(RabbitComsumer.run, items)

 

这篇关于python 使用pika对接rabbitMQ的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---