Rabbitmq - 在同一频道上消费和发布？答案

【问题标题】：Rabbitmq - consuming and publishing on same channel?Rabbitmq - 在同一频道上消费和发布？
【发布时间】：2014-07-31 22:34:47
【问题描述】：

我有两个通过 pika 连接到 Rabbitmq 的 python 进程。每个都使用一组主题，对方将这些主题作为响应发布。一种使用 SelectConnection，另一种使用 TornadoConnection。

目前这两个程序都只是模拟用户和我的服务器之间的对话的测试程序，每个程序的 on_message() 只是简单地硬编码以在收到的 routing_key 上进行分支并将适当的响应发布到它的对应方。

最初，经过一段随机时间（通常不超过 2 分钟）后，我会收到如下错误：

UnexpectedFrame: Basic.publish: (505) UNEXPECTED_FRAME - expected content header for class 60, got non content header frame instead

在此处搜索有关堆栈溢出和其他地方的大量帖子后，我了解到此错误与在 basic_publish 完成之前消耗某些东西的竞争条件有关。

我对我的代码进行了更改，因此我没有立即执行 basic_publish()，而是将回调传递给 connection.add_timeout()，延迟为 1 秒。进行此更改后，我已经能够进行多次运行，其中两个进程彼此“对话”超过 1 小时而不会重现错误。

我的问题是，这仅仅是因为我正在模拟一个用户而起作用的黑客行为吗？我需要有 2 个独立的消费和发布渠道吗？

def on_message(self, unused_channel, basic_deliver, properties, body):
    if self._sibling_app_id == properties.app_id:
        self.dispatch_message(basic_deliver, properties, body)


def dispatch_message(self, basic_deliver, properties, body):
    (user_id, msg_type) = basic_deliver.routing_key.rsplit('.', 1)

    if "login-response" == msg_type:
        print body
    elif "gid-assignment" == msg_type:
        print body
    elif "tutor-logout" == msg_type:
        print body
    elif "tutor-turn" == msg_type:
        message = "i don't know"
        routing_key = "%s.input" % user_id
        callback = self.delayed_publish_message(routing_key, message)
        self.schedule_next_message(callback, 1)
    elif "nlu" == msg_type:
        message = "dnk"
        routing_key = "%s.nlu-response" % user_id
        callback = self.delayed_publish_message(routing_key, message)
        self.schedule_next_message(callback, 1)
    else:
        print "invalid message-type: %s" % msg_type
        print body

def delayed_publish_message(self, routing_key, message):
    """returns a callback which can be passed to schedule_next_message()"""
    def delayed_publish_cb():
        self.publish_message(routing_key, message)
    return delayed_publish_cb


def schedule_next_message(self, cb, publish_interval=None):
    if self._stopping:
        return
    if publish_interval is None:
        publish_interval = self.PUBLISH_INTERVAL
    if -1 == publish_interval:
        return
    self._connection.add_timeout(publish_interval, cb)


def publish_message(self, routing_key, message):
    if self._stopping:
        return
    properties = pika.BasicProperties(app_id=self._app_id,
                                                          content_type='text/plain')
    self._channel.basic_publish(self.EXCHANGE, routing_key,
                                                 message, properties)

【问题讨论】：

哪个进程抛出了异常，是SelectConnection还是TornadoConnection？
总是选择连接
有趣。根据this bug report，与SelectConnection 相同的错误已在2012 年与pull request 一起修复。我不确定这是否是一种回归（基于对当前来源的快速浏览，它看起来不像） , 类似的问题有不同的原因，或者如果原始修复错过了某些情况。
是的，我注意到在我的搜索中提到了该特定修复。这是否意味着我的代码应该可以工作，即使没有添加 1 秒的延迟？
@dano 感谢您的 cmets，虽然我之前注意到了拉取请求，但我没有注意到数据。如果你没有提到这一点，我可能不会想到这一点

标签： python python-2.7 rabbitmq pika

【解决方案1】：

通道将被单向使用。 AMQP protocol specification 对此非常清楚：

一个 AMQP 会话将两个单向通道关联起来，以在两个容器之间形成一个双向的顺序会话。一种单个连接可能有多个独立的活动会话同时，达到协商的通道限制。两个连接会话由每个对等方建模为存储本地的端点以及关于连接或会话的最后已知远程状态问题。

因此，您应该为您的应用程序使用输入和输出通道。

【讨论】：

您引用了 AMQP 1.0 规范，但 RabbitMQ 实现了 0.9.1 版。 0.9.1 规范中是否有类似的条款？
0.9.1 规范没有解释 AMQP 会话。但是第 4.7 节（内容订购保证）中有一个提示。另一方面：通道实际上是一种哑线。两边同时说话不是一个好主意，会导致帧错误。

【解决方案2】：

我做了我的承诺，正要上床睡觉，突然我想通了。我发现 rabbitmq.com 上的 python 教程仍然说要安装 pika：

 sudo pip install pika==0.9.8

虽然 0.9.8 于 2012 年某个时候发布，但我认为修复是在该版本发布之后添加的。 0.9.9 将于 2013 年某个时候发布

所以，我做到了：

sudo pip uninstall pika

按照 pika 网站上的安装说明进行操作：

sudo pip install pika

然后我用 basic_publish() 替换了所有的 connection.add_timeout(1,delayed_publish_cb)，交叉手指，运行它，我的两个进程在不到 5 分钟的时间内相互交换了大约 200,000 条消息，没有任何问题

很高兴知道 2012 年的错误修复仍然有效。

我必须让 rabbitmq 人知道更新他们的教程。

【讨论】：