如果消息流超出了该用户可以处理的范围,则我有一个用户可以监听消息,我想启动该用户的另一个实例.

但是我还希望能够轮询来自消费者的信息,我的想法是我可以使用扇出交换,使用RPC向生产者请求此信息,以便所有生产者都可以进行RPC调用.

我的问题是,首先这是否可能,其次是否合理？

解决方法:

如果问题是“是否可以将RPC消息发送到多个服务器？”答案是肯定的.

构建RPC调用时,您会在消息上附加一个临时队列(通常在header.reply_to中,但您也可以使用内部消息字段).这是RPC目标将发布其答案的队列.

当您将RPC发送到单个服务器时,您可以在临时队列上收到多个消息：这意味着RPC答案可以由以下方式形成：

>来自单一来源的单一消息
>来自单一来源的多条消息
>来自多个来源的多个消息

在这种情况下出现的问题是

>什么时候停止收听？如果您知道RPC服务器的数量,则可以等到每个服务器都向您发送答案为止,否则您必须实施某种形式的超时
>您是否需要跟踪答案的来源？您可以在邮件中添加一些特殊字段以保留此信息.消息顺序相同.

只需一些代码即可演示如何实现(带有Pika库的Python).请注意,这远非完美：最大的问题是,当您获得新答案时应重置超时.

    def consume_rpc(self, queue, result_len=1, callback=None, timeout=None, raise_timeout=False):
        if timeout is None:
            timeout = self.rpc_timeout

        result_list = []

        def _callback(channel, method, header, body):
            print "### Got 1/%s RPC result" %(result_len)
            msg = self.encoder.decode(body)
            result_dict = {}
            result_dict.update(msg['content']['data'])
            result_list.append(result_dict)

            if callback is not None:
                callback(msg)

            if len(result_list) == result_len:
                print "### All results are here: stopping RPC"
                channel.stop_consuming()

        def _outoftime():
            self.channel.stop_consuming()
            raise TimeoutError

        if timeout != -1:
            print "### Setting timeout %s seconds" %(timeout)
            self.conn_broker.add_timeout(timeout, _outoftime)

        self.channel.basic_consume(_callback, queue=queue, consumer_tag=queue)

        if raise_timeout is True:
            print "### Start consuming RPC with raise_timeout"
            self.channel.start_consuming()
        else:
            try:
                print "### Start consuming RPC without raise_timeout"
                self.channel.start_consuming()
            except TimeoutError:
                pass

        return result_list