RabbitMQ-入门

"""
整体实现基于同步模式：

- BlockingConnection 阻塞/连接

> 该适配器虽然是同步阻塞适配器，但是依然支持异步RPC特性
> 组成： BlockingConnection 和 BlockingChannel
> 异步适配器
> 1.Select Connection Adapter 没有第三方依赖包的异步模式
> 2.Tornado Connection Adapter 基于Tornado 的异步IO请求模式
> 3.Twisted Connection Adapter 基于Twisted’的异步IO请求模式
"""
# coding:utf8
from common.config.dev import RABBITMQ_HOST,RABBITMQ_PASS,RABBITMQ_PORT,RABBITMQ_USER
import uuid
import pika
import json
from common.utils.logger import Logger
from common.utils.helpers import JsonObject
logger = Logger()



class RabbitMQ():
    def __init__(self,v_host="/",host=RABBITMQ_HOST, port=RABBITMQ_PORT, user=RABBITMQ_USER, password=RABBITMQ_PASS):
        """
        RabbitMQ 封装
        :param v_host: 虚拟机host
        :param host: RabbitMQ 服务host
        :param port: RabbitMQ服务端口
        :param user: RabbitMq 用户认证-用户名
        :param password: RabbitMq 用户认证-密码
        """
        self._v_host = v_host
        self._host = host
        self._port = port
        self._user = user
        self._password = password
        self.exchange_type = ""
        self.exchange = ""
        self.responses = {} #rpc客户端收到的所有结果
        self.do_connection()

    def do_connection(self):
        """
        进行了连接这个动作
        :return:
        """
        try:
            self.connection = self.blocking_connection(host=self._host,
                                     port=self._port,
                                     user=self._user,
                                     password=self._password,
                                    v_host=self._v_host)
            self.channel = self.connection.channel()
        except ValueError as e:
            print(e)

    def blocking_connection(self,
                            host=RABBITMQ_HOST,
                            port=RABBITMQ_PORT,
                            user=RABBITMQ_USER,
                            password=RABBITMQ_PASS,
                            v_host="/"):
        """
        Connection Adapters 连接适配器
        pika.PlainCredentials(
        ：param str用户名：用于验证的用户名,默认为guest
        ：param str password：用于验证的密码，默认为guest
        ：param bool delete_on_connect：连接完成后，凭证将不会存储在内存中。删除连接凭据。
        )
        pika.ConnectionParameters(
        host=host
        port=port,
        virtual_host=要使用的rabbitmq虚拟主机, 一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mq
        credentials=PlainCredentials身份验证凭证,
        channel_max=允许的最大channel通道数,
        frame_max=AMQP frame的最大字节大小,
        heartbeat=AMQP callable 心跳超时协商,
        ssl_options=pika.SSLOptions`实例,
        connection_attempts=最大重试尝试次数,
        retry_delay=下一次等待的时间，以秒为单位,
        socket_timeout=套接字连接超时时间,
        stack_timeout=完整协议栈启动超时时间，应该高于socket_timeout,
        locale=设置语言环境值,
        blocked_connection_timeout=如果非负，阻止链接保持；如果
        tcp_options=为套接字设置的TCP选项的字典
        ）
        """
        credential_params = pika.PlainCredentials(user, password)
        connection = pika.BlockingConnection(
            pika.ConnectionParameters(host, port, v_host, credential_params)
        )
        return connection

    def model(self,exchange,model="fanout"):
        """
        可以选择，自己想构建的模式，
        然后该函数会把需要设置的交换机参数帮你设置好
        :param model:
            model_exchangeType = {
                "fanout":"fanout",
                "worker":"fanout",
                "routing":"direct",
                "direct":"direct",
                "route":"direct",
                "topic":"topic",
                "rpc":"direct"#默认为路由模式
            }
        :return:
        """
        model_exchangeType = {
            "fanout": "fanout",
            "worker": "fanout",
            "routing": "direct",
            "direct": "direct",
            "route": "direct",
            "topic": "topic",
            "rpc": "direct"
        }
        self.exchange_type = model_exchangeType[model]
        self.exchange = exchange
        return self

    def producer_init(self,exchange=None,exchange_type=None):
        """
        生产者初始化代码
        :param v_host: 默认为/，默认虚拟机
        :param exchange:交换机名字，默认为“”
        :param exchange_type:交换机类型，默认为“fanout”,广播模式
        :return: channel
        """
        try:
            self.exchange = exchange if exchange else self.exchange
            self.exchange_type = exchange_type if exchange_type else self.exchange_type
            self.check_Value(self.exchange, self.exchange_type, self.channel)
            """默认这里connection 和 channel都有了，没有raise 异常"""
            # 声明交换机
            self.channel.exchange_declare(exchange=self.exchange, exchange_type=self.exchange_type)
            return self
        except Exception as e:
            print(e)

    def consumer_init(self,exchange=None,exchange_type=None,queue="",routings=[""]):
        """
        消费者初始化代码
        :param exchange: 交换机名字，默认为
        :param exchange_type: 交换机类型，默认为“fanout”，广播模式
        :param queue: 队列名，默认为“”，为随机队列
        :param routings: 路由key列表，默认为[""],会将这些路由key都绑定到该队列上
        :return: channel
        """
        try:
            self.exchange = exchange if exchange else self.exchange
            self.exchange_type = exchange_type if exchange_type else self.exchange_type
            self.check_Value(self.exchange,self.exchange_type,self.channel)
            """默认这里connection 和 channel都有了，没有raise 异常"""
            # 声明交换机
            self.channel.exchange_declare(exchange=self.exchange, exchange_type=self.exchange_type)
            # 随机生成临时队列,绑定到交换机.当然也可以指定出多条名字固定的队列，exclusive设置为True当接收端退出时，自动销毁队列
            self.queue = self.channel.queue_declare(queue=queue, auto_delete=True)
            self.qname = self.queue.method.queue
            # 将这个随机队列绑定到交换机上，并设置路由键
            for routing in routings:
                self.channel.queue_bind(exchange=self.exchange,queue=self.qname,routing_key=routing)
            return self
        except Exception as e:
            print(e)

    def rpc_client_init(self,routings:list,exchange=None,exchange_type=None,queue="",auto_ack=True):
        """
        rpc 模型
        1. 先定义消费者
        2. 再定义生产者
        :param exchange: 客户端接受调用结果的交换机
        :param exchange_type: 客户端接受调用结果的交换机的类型
        :param queue: 客户端接受调用结果的队列，当为空时，该队列名字随机
        :param routings: 客户端接受调用结果的队列路由键,必选参数，客户端接受调用结果的队列的路由
        :param auto_ack: 是否自动确认
        :return:
        """
        try:
            self.consumer_init(exchange=exchange, exchange_type=exchange_type, queue=queue, routings=routings)
            self.channel.basic_consume(queue=self.qname, auto_ack=auto_ack, on_message_callback=self.rpc_client_response)
            return self
        except Exception as e:
            print(e)

    def rpc_client_response(self,ch, method, properties, body)->any:
        """
        收到结果后的回调函数
        :param ch:
        :param method:
        :param properties:
        :param body:
        :return:
        """
        if self.corr_id == properties.correlation_id:
            self.response = json.loads(body)

    def rpc_client_request(self,body:any,exchange:str,routing_key:str):
        """
        启动rpc客户端请求循环，拿到消息后返回
        :param body: 请求体
        :param exchange:服务端的交换机
        :param routing_key:请求发送到服务端的哪个队列，由路由指定
        :return:
        """

        self.response = None
        self.corr_id = str(uuid.uuid4())
        self.responses[self.corr_id] = None
        self.channel.basic_publish(exchange=exchange,
                                   routing_key=routing_key,
                                   properties=pika.BasicProperties(
                                       reply_to=self.qname,
                                       correlation_id=self.corr_id
                                   ),
                                   body=json.dumps(body))
        while self.response is None:
            self.connection.process_data_events()
        self.responses[self.corr_id] = self.response
        return self.response

    def rpc_server_init(self,call=None,exchange=None,exchange_type=None,queue="",routings=[""]):
        """
        rpc服务端初始化，会初始化一条消费者队列，这个队列的到达路由需要和客户端进行约定
        :param call: 处理远程调用请求的函数
        :param exchange: 服务端的客户端约定好的交换机
        :param exchange_type: 交换机类型
        :param queue: 队列名，默认为""，为临时队列，名字随机
        :param routings:路由键，服务端的队列需要绑定到某个路由上，默认为"",即该交换机为广播模式时生效
        :return:self
        """
        try:
            #将可以主动设置处理远程调用请求的函数，也可以由回调函数rpc_server_callback自己去找
            self.remote_callback = call
            self.consumer_init(exchange=exchange,exchange_type=exchange_type,queue=queue,routings=routings)
            self.channel.basic_qos(prefetch_count=100)
            self.channel.basic_consume(queue=self.qname, auto_ack=False,on_message_callback=self.rpc_server_callback)
            return self
        except Exception as e:
            print(e)

    def rpc_server_callback(self,ch, method, properties, body):
        """
        这里应该是解析参数，寻找被调用者，然后进行调用后
        将调用结果发送会请求中要求返回的路由中
        :param ch: channel
        :param method: method
        :param properties: 客户端发的消息带的参数
        :param body: 请求体，会封装调用信息
        :return:
        """
        # 将计算结果发送回该请求体要求发送的路由中
        # 这里的发送，直接发送到请求体传过来的队列中，所以设置exchange=""
        ch.basic_publish(exchange="",
                         routing_key=properties.reply_to,
                         properties=pika.BasicProperties(
                             reply_to=self.qname,
                             correlation_id=properties.correlation_id
                         ),
                         body=json.dumps(self.remote_callback(json.loads(body))),
                         )
        ch.basic_ack(delivery_tag=method.delivery_tag)

    def rpc_server_start(self):
        """
        启动rpc的服务端
        :return:
        """
        self.channel.start_consuming()

    def check_Value(self,*args):
        """参数校验"""
        for i in args:
            assert i != None