pika详解(三)SelectConnection及其他Connection

我会带着你远行 2021-12-14 12:07 446阅读 0赞

pika除了block connection外还提供了其他非阻塞式的连接方式

**SelectConnection**

定义如下：

class SelectConnection(BaseConnection):
        def __init__(
                self,  # pylint: disable=R0913
                parameters=None,
                on_open_callback=None,
                on_open_error_callback=None,
                on_close_callback=None,
                custom_ioloop=None,
                internal_connection_workflow=True):
            """Create a new instance of the Connection object.
    
            :param pika.connection.Parameters parameters: Connection parameters
            :param callable on_open_callback: Method to call on connection open
            :param None | method on_open_error_callback: Called if the connection
                can't be established or connection establishment is interrupted by
                `Connection.close()`: on_open_error_callback(Connection, exception).
            :param None | method on_close_callback: Called when a previously fully
                open connection is closed:
                `on_close_callback(Connection, exception)`, where `exception` is
                either an instance of `exceptions.ConnectionClosed` if closed by
                user or broker or exception of another type that describes the cause
                of connection failure.
            :param None | IOLoop | nbio_interface.AbstractIOServices custom_ioloop:
                Provide a custom I/O Loop object.
            :param bool internal_connection_workflow: True for autonomous connection
                establishment which is default; False for externally-managed
                connection workflow via the `create_connection()` factory.
            :raises: RuntimeError
    
            """
            if isinstance(custom_ioloop, nbio_interface.AbstractIOServices):
                nbio = custom_ioloop
            else:
                nbio = SelectorIOServicesAdapter(custom_ioloop or IOLoop())
    
            super(SelectConnection, self).__init__(
                parameters,
                on_open_callback,
                on_open_error_callback,
                on_close_callback,
                nbio,
                internal_connection_workflow=internal_connection_workflow)
    
        @classmethod
        def create_connection(cls,
                              connection_configs,
                              on_done,
                              custom_ioloop=None,
                              workflow=None):
            """Implement
            :py:classmethod:`pika.adapters.BaseConnection.create_connection()`.
    
            """
            nbio = SelectorIOServicesAdapter(custom_ioloop or IOLoop())
    
            def connection_factory(params):
                """Connection factory."""
                if params is None:
                    raise ValueError('Expected pika.connection.Parameters '
                                     'instance, but got None in params arg.')
                return cls(
                    parameters=params,
                    custom_ioloop=nbio,
                    internal_connection_workflow=False)
    
            return cls._start_connection_workflow(
                connection_configs=connection_configs,
                connection_factory=connection_factory,
                nbio=nbio,
                workflow=workflow,
                on_done=on_done)

SelectConnection继承BaseConnection，只是重写了create\_connection

在create\_connection 中可以看到：

nbio = SelectorIOServicesAdapter(custom\_ioloop or IOLoop())

说明这个类实现的 select IO

提供的函数、方法解析：

1.  add\_on\_close\_callback

关闭时候的回调， callback定义 callback（pika.connection.Connection，exception）

1.  add\_on\_connection\_blocked\_callback

阻塞时候回调 回调定义： callback(connection, pika.frame.Method)

1.  add\_on\_connection\_unblocked\_callback
    
    解除阻塞时候回调
2.  add\_on\_open\_callback
    
    打开时回调
3.  add\_on\_open\_error\_callback

打开异常时候回调

1.  basic\_nack
2.  channel（channel\_number = None\*，\*on\_open\_callback = None **）**

建立通道

channel\_number（int） - 要使用的通道编号，默认为下一个可用通道编号。  
on\_open\_callback（callable）

1.  close
2.  consumer\_cancel\_notify

定服务器是否支持活动连接上的使用者取消通知 返回bool值

1.  create\_connection(connection\_configs, on\_done, custom\_ioloop=None, workflow=None)

建立连接

1.  exchange\_exchange\_bindings
    
    返回值bool 是否支持交换机修改
2.  ioloop
3.  is\_closed
4.  is\_closing
5.  is\_open
6.  publisher\_confirms
    
    是否支持发布确认

Tornado Connection Adapter

方法可参考基类BaseConnection

略

Twisted Connection Adapter

方法可参考基类BaseConnection

略