aio-pika 2 消费者不是在同一时间通过扇出接收消息答案

【问题标题】：aio-pika 2 consumers receive message through fanout not in the same timeaio-pika 2 消费者不是在同一时间通过扇出接收消息
【发布时间】：2019-12-15 00:27:29
【问题描述】：

我是 RabbitMQ 和 Pika 的新手，但我认为我已经清楚地了解它是如何工作的。

我需要实现这个：

生产者创建消息并通过扇出交换发送，多个生产者（测试环境中的两个）接收相同的消息。

但一次只有 1 个消费者收到消息

2019-11-29 19:02:44.167549 b'Hello' - 第一个消费者

2019-11-29 19:02:45.068192 b'Hello' - 第二个消费者

制作人：

    async def main(loop):
        connection = await connect_robust(
            "amqp://guest:guest@192.168.1.3/", loop=loop
        )

        queue_name = "test_queue"
        routing_key = "test_queue"

        # Creating channel
        channel = await connection.channel()

        # Declaring exchange
        exchange = await channel.declare_exchange('test_exchange',
                                                  ExchangeType.FANOUT, auto_delete=True
                                                  )

        # Declaring queue
        queue = await channel.declare_queue(
            queue_name, auto_delete=True
        )

        # Binding queue
        await queue.bind(exchange, routing_key)

        await exchange.publish(
            Message(
                bytes('Hello', 'utf-8'),
                content_type='text/plain',
                headers={'foo': 'bar'}
            ),
            routing_key
        )
    )


    if __name__ == "__main__":
        loop = asyncio.get_event_loop()
        loop.run_until_complete(main(loop))

消费者：

    async def main(loop):
            connection = await aio_pika.connect_robust(host='192.168.1.3', login='guest', password='guest', loop=loop
                                                       )

            queue_name = "test_queue"

            async with connection:
                # Creating channel
                channel = await connection.channel()

                # Declaring queue
                queue = await channel.declare_queue(
                    queue_name, auto_delete=True
                )

                async with queue.iterator() as queue_iter:
                    async for message in queue_iter:
                        async with message.process():
                            print(datetime.datetime.now(), message.body)

                            if queue.name in message.body.decode():
                                break


    if __name__ == "__main__":
          loop = asyncio.get_event_loop()
          loop.run_until_complete(main(loop))
          loop.close()

【问题讨论】：

标签： python rabbitmq pika

【解决方案1】：

首先，我假设您正在运行两个独立的消费者进程。

每个消费者都应该将自己的队列绑定到扇出交换。不要使用共享队列。一种解决方案是让每个消费者使用独占队列。

只要您的消费者先开始，生产者就无需创建队列并将其绑定到扇出交换。

先试试这个。然后，如果您需要考虑到您的生产者可以先启动，它必须创建两个具有众所周知的名称的队列，并绑定它们。消费者在开始时应该做同样的事情。

_{注意：RabbitMQ 团队会监控 rabbitmq-users mailing list 并且有时只回答 StackOverflow 上的问题。}

【讨论】：