rabbitmq

RabbitMQ 消息队列。

RabbitMQMessageQueueConfig

基类: BaseSettings

RabbitMQ 消息队列配置。

参数

名称	类型	描述	默认值
type	Literal[str]		'rabbitmq'
url	str		'amqp://guest:guest@localhost/'
exchange_name	str		'llama-deploy'
username	str | None		无
password	str | None		无
host	str | None		无
port	int | None		无
vhost	str | None		无
secure	bool | None		无

源代码位于 llama_deploy/message_queues/rabbitmq.py 中

class RabbitMQMessageQueueConfig(BaseSettings):
    """RabbitMQ message queue configuration."""

    model_config = SettingsConfigDict(env_prefix="RABBITMQ_")

    type: Literal["rabbitmq"] = Field(default="rabbitmq", exclude=True)
    url: str = DEFAULT_URL
    exchange_name: str = DEFAULT_EXCHANGE_NAME
    username: str | None = None
    password: str | None = None
    host: str | None = None
    port: int | None = None
    vhost: str | None = None
    secure: bool | None = None

    def model_post_init(self, __context: Any) -> None:
        if self.username and self.password and self.host:
            scheme = "amqps" if self.secure else "amqp"
            self.url = f"{scheme}://{self.username}:{self.password}@{self.host}"
            if self.port:
                self.url += f":{self.port}"
            elif self.vhost:
                self.url += f"/{self.vhost}"

RabbitMQMessageQueue

基类: AbstractMessageQueue

使用 aio-pika 客户端集成 RabbitMQ。

此类创建一个工作（或任务）队列。有关 RabbitMQ 工作队列的更多信息，请参阅下方链接的页面： 1. https://aio-pika.readthedocs.io/en/stable/rabbitmq-tutorial/2-work-queues.html 2. https://aio-pika.readthedocs.io/en/stable/rabbitmq-tutorial/3-publish-subscribe.html

连接通过使用 amqp uri scheme 的 url 建立

amqp_URI       = "amqp://" amqp_authority [ "/" vhost ] [ "?" query ]
amqp_authority = [ amqp_userinfo "@" ] host [ ":" port ]
amqp_userinfo  = username [ ":" password ]
username       = *( unreserved / pct-encoded / sub-delims )
password       = *( unreserved / pct-encoded / sub-delims )
vhost          = segment

创建的工作队列具有以下属性

• 名称为 self.exchange 的交换机
• 消息通过此交换机发布到此队列
• 消费者绑定到交换机，并根据其消息类型拥有自己的队列
• 轮询分发：当多个消费者监听同一队列时，只有一个消费者会被选中，由顺序决定。

属性

名称	类型	描述
url	str	连接到 RabbitMQ 服务器的 amqp url 字符串
exchange_name	str	在 RabbitMQ AMQP 0-9 协议中给“交换机”起的名称。

示例

from llama_deploy.message_queues.rabbitmq import RabbitMQMessageQueue

message_queue = RabbitMQMessageQueue()  # uses the default url

源代码位于 llama_deploy/message_queues/rabbitmq.py 中

class RabbitMQMessageQueue(AbstractMessageQueue):
    """RabbitMQ integration with aio-pika client.

    This class creates a Work (or Task) Queue. For more information on Work Queues
    with RabbitMQ see the pages linked below:
        1. https://aio-pika.readthedocs.io/en/stable/rabbitmq-tutorial/2-work-queues.html
        2. https://aio-pika.readthedocs.io/en/stable/rabbitmq-tutorial/3-publish-subscribe.html

    Connections are established by url that use [amqp uri scheme](https://rabbitmq.cn/docs/uri-spec#the-amqp-uri-scheme):

    ```
    amqp_URI       = "amqp://" amqp_authority [ "/" vhost ] [ "?" query ]
    amqp_authority = [ amqp_userinfo "@" ] host [ ":" port ]
    amqp_userinfo  = username [ ":" password ]
    username       = *( unreserved / pct-encoded / sub-delims )
    password       = *( unreserved / pct-encoded / sub-delims )
    vhost          = segment
    ```

    The Work Queue created has the following properties:
        - Exchange with name self.exchange
        - Messages are published to this queue through the exchange
        - Consumers are bound to the exchange and have queues based on their
            message type
        - Round-robin dispatching: with multiple consumers listening to the same
            queue, only one consumer will be chosen dictated by sequence.

    Attributes:
        url (str): The amqp url string to connect to the RabbitMQ server
        exchange_name (str): The name to give to the so-called exchange within
            RabbitMQ AMQP 0-9 protocol.

    Examples:
        ```python
        from llama_deploy.message_queues.rabbitmq import RabbitMQMessageQueue

        message_queue = RabbitMQMessageQueue()  # uses the default url
        ```
    """

    def __init__(
        self,
        config: RabbitMQMessageQueueConfig | None = None,
        url: str = DEFAULT_URL,
        exchange_name: str = DEFAULT_EXCHANGE_NAME,
        **kwargs: Any,
    ) -> None:
        self._config = config or RabbitMQMessageQueueConfig()
        self._registered_topics: set[str] = set()

    @classmethod
    def from_url_params(
        cls,
        username: str,
        password: str,
        host: str,
        vhost: str = "",
        port: int | None = None,
        secure: bool = False,
        exchange_name: str = DEFAULT_EXCHANGE_NAME,
    ) -> "RabbitMQMessageQueue":
        """Convenience constructor from url params.

        Args:
            username (str): username for the amqp authority
            password (str): password for the amqp authority
            host (str): host for rabbitmq server
            port (int | None, optional): port for rabbitmq server. Defaults to None.
            secure (bool, optional): Whether or not to use SSL. Defaults to False.
            exchange_name (str, optional): The exchange name. Defaults to DEFAULT_EXCHANGE_NAME.

        Returns:
            RabbitMQMessageQueue: A RabbitMQ MessageQueue integration.
        """
        if not secure:
            if port:
                url = f"amqp://{username}:{password}@{host}:{port}/{vhost}"
            else:
                url = f"amqp://{username}:{password}@{host}/{vhost}"
        else:
            if port:
                url = f"amqps://{username}:{password}@{host}:{port}/{vhost}"
            else:
                url = f"amqps://{username}:{password}@{host}/{vhost}"
        return cls(RabbitMQMessageQueueConfig(url=url, exchange_name=exchange_name))

    async def new_connection(self) -> "Connection":
        """Returns a new connection to the RabbitMQ server."""
        return await _establish_connection(self._config.url)

    async def _publish(self, message: QueueMessage, topic: str) -> Any:
        """Publish message to the queue."""
        from aio_pika import DeliveryMode, ExchangeType
        from aio_pika import Message as AioPikaMessage

        connection = await _establish_connection(self._config.url)

        async with connection:
            channel = await connection.channel()
            exchange = await channel.declare_exchange(
                self._config.exchange_name,
                ExchangeType.DIRECT,
            )
            message_body = json.dumps(message.model_dump()).encode("utf-8")

            aio_pika_message = AioPikaMessage(
                message_body,
                delivery_mode=DeliveryMode.PERSISTENT,
            )
            # Sending the message
            await exchange.publish(aio_pika_message, routing_key=topic)
            logger.info(f"published message {message.id_} to {topic}")

    async def register_consumer(
        self, consumer: BaseMessageQueueConsumer, topic: str
    ) -> StartConsumingCallable:
        """Register a new consumer."""
        from aio_pika import Channel, ExchangeType, IncomingMessage, Queue
        from aio_pika.abc import AbstractIncomingMessage

        connection = await _establish_connection(self._config.url)
        async with connection:
            channel = cast(Channel, await connection.channel())
            exchange = await channel.declare_exchange(
                self._config.exchange_name,
                ExchangeType.DIRECT,
            )
            queue = cast(Queue, await channel.declare_queue(name=topic))
            await queue.bind(exchange)

        self._registered_topics.add(topic or consumer.message_type)
        logger.info(
            f"Registered consumer {consumer.id_} for topic: {topic}",
        )

        async def start_consuming_callable() -> None:
            """StartConsumingCallable.

            Consumer of this queue, should call this in order to start consuming.
            """

            async def on_message(message: AbstractIncomingMessage) -> None:
                message = cast(IncomingMessage, message)
                async with message.process():
                    decoded_message = json.loads(message.body.decode("utf-8"))
                    queue_message = QueueMessage.model_validate(decoded_message)
                    await consumer.process_message(queue_message)

            # The while loop will reconnect if the connection is lost
            while True:
                connection = await _establish_connection(self._config.url)
                try:
                    async with connection:
                        channel = await connection.channel()
                        exchange = await channel.declare_exchange(
                            self._config.exchange_name,
                            ExchangeType.DIRECT,
                        )
                        queue = cast(Queue, await channel.declare_queue(name=topic))
                        await queue.bind(exchange)
                        await queue.consume(on_message)
                        await asyncio.Future()
                except asyncio.CancelledError:
                    logger.info(
                        f"Cancellation requested, exiting consumer {consumer.id_} for topic: {topic}"
                    )
                    break
                except Exception as e:
                    logger.error(f"Unexpected error: {e}", exc_info=True)
                    # Wait before reconnecting. Ideally we'd want exponential backoff here.
                    await asyncio.sleep(10)
                finally:
                    if connection:
                        await connection.close()

        return start_consuming_callable

    async def deregister_consumer(self, consumer: BaseMessageQueueConsumer) -> Any:
        """Deregister a consumer.

        Not implemented for this integration, as once the connection/channel is
        closed, the consumer is deregistered.
        """
        pass

    async def cleanup(self, *args: Any, **kwargs: dict[str, Any]) -> None:
        """Perform any clean up of queues and exchanges."""
        connection = await self.new_connection()
        async with connection:
            channel = await connection.channel()
            for queue_name in self._registered_topics:
                await channel.queue_delete(queue_name=queue_name)
            await channel.exchange_delete(exchange_name=self._config.exchange_name)

    def as_config(self) -> BaseModel:
        return RabbitMQMessageQueueConfig(
            url=self._config.url, exchange_name=self._config.exchange_name
        )

from_url_params classmethod

from_url_params(username: str, password: str, host: str, vhost: str = '', port: int | None = None, secure: bool = False, exchange_name: str = DEFAULT_EXCHANGE_NAME) -> RabbitMQMessageQueue

从 url 参数创建的便利构造函数。

参数

名称	类型	描述	默认值
username	str	amqp 授权的用户名	必需
password	str	amqp 授权的密码	必需
host	str	rabbitmq 服务器的主机	必需
port	int | None	rabbitmq 服务器的端口。默认为 None。	无
secure	bool	是否使用 SSL。默认为 False。	False
exchange_name	str	交换机名称。默认为 DEFAULT_EXCHANGE_NAME。	DEFAULT_EXCHANGE_NAME

返回

名称	类型	描述
RabbitMQMessageQueue	RabbitMQMessageQueue	一个 RabbitMQ MessageQueue 集成。

源代码位于 llama_deploy/message_queues/rabbitmq.py 中

@classmethod
def from_url_params(
    cls,
    username: str,
    password: str,
    host: str,
    vhost: str = "",
    port: int | None = None,
    secure: bool = False,
    exchange_name: str = DEFAULT_EXCHANGE_NAME,
) -> "RabbitMQMessageQueue":
    """Convenience constructor from url params.

    Args:
        username (str): username for the amqp authority
        password (str): password for the amqp authority
        host (str): host for rabbitmq server
        port (int | None, optional): port for rabbitmq server. Defaults to None.
        secure (bool, optional): Whether or not to use SSL. Defaults to False.
        exchange_name (str, optional): The exchange name. Defaults to DEFAULT_EXCHANGE_NAME.

    Returns:
        RabbitMQMessageQueue: A RabbitMQ MessageQueue integration.
    """
    if not secure:
        if port:
            url = f"amqp://{username}:{password}@{host}:{port}/{vhost}"
        else:
            url = f"amqp://{username}:{password}@{host}/{vhost}"
    else:
        if port:
            url = f"amqps://{username}:{password}@{host}:{port}/{vhost}"
        else:
            url = f"amqps://{username}:{password}@{host}/{vhost}"
    return cls(RabbitMQMessageQueueConfig(url=url, exchange_name=exchange_name))

new_connection async

new_connection() -> Connection

返回与 RabbitMQ 服务器的新连接。

源代码位于 llama_deploy/message_queues/rabbitmq.py 中

async def new_connection(self) -> "Connection":
    """Returns a new connection to the RabbitMQ server."""
    return await _establish_connection(self._config.url)

register_consumer async

register_consumer(consumer: BaseMessageQueueConsumer, topic: str) -> StartConsumingCallable

注册一个新消费者。

源代码位于 llama_deploy/message_queues/rabbitmq.py 中

async def register_consumer(
    self, consumer: BaseMessageQueueConsumer, topic: str
) -> StartConsumingCallable:
    """Register a new consumer."""
    from aio_pika import Channel, ExchangeType, IncomingMessage, Queue
    from aio_pika.abc import AbstractIncomingMessage

    connection = await _establish_connection(self._config.url)
    async with connection:
        channel = cast(Channel, await connection.channel())
        exchange = await channel.declare_exchange(
            self._config.exchange_name,
            ExchangeType.DIRECT,
        )
        queue = cast(Queue, await channel.declare_queue(name=topic))
        await queue.bind(exchange)

    self._registered_topics.add(topic or consumer.message_type)
    logger.info(
        f"Registered consumer {consumer.id_} for topic: {topic}",
    )

    async def start_consuming_callable() -> None:
        """StartConsumingCallable.

        Consumer of this queue, should call this in order to start consuming.
        """

        async def on_message(message: AbstractIncomingMessage) -> None:
            message = cast(IncomingMessage, message)
            async with message.process():
                decoded_message = json.loads(message.body.decode("utf-8"))
                queue_message = QueueMessage.model_validate(decoded_message)
                await consumer.process_message(queue_message)

        # The while loop will reconnect if the connection is lost
        while True:
            connection = await _establish_connection(self._config.url)
            try:
                async with connection:
                    channel = await connection.channel()
                    exchange = await channel.declare_exchange(
                        self._config.exchange_name,
                        ExchangeType.DIRECT,
                    )
                    queue = cast(Queue, await channel.declare_queue(name=topic))
                    await queue.bind(exchange)
                    await queue.consume(on_message)
                    await asyncio.Future()
            except asyncio.CancelledError:
                logger.info(
                    f"Cancellation requested, exiting consumer {consumer.id_} for topic: {topic}"
                )
                break
            except Exception as e:
                logger.error(f"Unexpected error: {e}", exc_info=True)
                # Wait before reconnecting. Ideally we'd want exponential backoff here.
                await asyncio.sleep(10)
            finally:
                if connection:
                    await connection.close()

    return start_consuming_callable

deregister_consumer async

deregister_consumer(consumer: BaseMessageQueueConsumer) -> Any

注销一个消费者。

对于此集成未实现，因为连接/通道关闭后，消费者即被注销。

源代码位于 llama_deploy/message_queues/rabbitmq.py 中

async def deregister_consumer(self, consumer: BaseMessageQueueConsumer) -> Any:
    """Deregister a consumer.

    Not implemented for this integration, as once the connection/channel is
    closed, the consumer is deregistered.
    """
    pass

cleanup async

cleanup(*args: Any, **kwargs: dict[str, Any]) -> None

执行队列和交换机的任何清理操作。

源代码位于 llama_deploy/message_queues/rabbitmq.py 中

async def cleanup(self, *args: Any, **kwargs: dict[str, Any]) -> None:
    """Perform any clean up of queues and exchanges."""
    connection = await self.new_connection()
    async with connection:
        channel = await connection.channel()
        for queue_name in self._registered_topics:
            await channel.queue_delete(queue_name=queue_name)
        await channel.exchange_delete(exchange_name=self._config.exchange_name)