go - rabbitmq/amqp 直接回复并返回通知

Question

我使用 rabbitmq 将来自各种输入源(websocket、rest、...)的消息传递给我的工作人员。每个工作人员在共享交换上监听一堆不同的路由 key 。

现在workerA有可能处理“routeA”。当我的输入源向 routeA 发送某些内容时，workerA 将其拾取并使用它。

但是如果没有 routeA-"consumer"会发生什么？在这种情况下，我希望输入源“知道”该请求没有人。并且由于没有“兔子消费者”消费此消息，因此将其丢弃。 (抱歉，如果术语不准确)据我了解控制消息(？)处理，这就是 NotifyReturn() (用于 amqp 的 golang 库)启动，以便发布者可以知道事实，它的消息被丢弃了。

这是我的代码的剥离示例。这种方法在一个简单的“仅发布此消息场景”中适用于我。我为 RPC 休息。
RPC 总是触发 case returnNotification := <-returnChannel:案子。

我的问题:

我拿错了吗？/这不是检查消息可传递性的方法吗？

谢谢!

编辑 :忘了提:“回复”中引发了错误。所以请求被发送，但是回复(通过“刚刚发布”发送)得到一个 NO_ROUTE returnNotification

刚发布

    // error handling omitted for example code
    tC, _ := rabbitConnection.Channel()
    defer tC.Close()

    tC.Confirm(false)

    var ack = make(chan uint64)
    var nack = make(chan uint64)
    tC.NotifyConfirm(ack, nack)

    returnChannel := make(chan amqp.Return)
    tC.NotifyReturn(returnChannel)

    p := someFunctionGeneratingAPublishing()

    tC.Publish(
        exchange,
        e.GetRoutingKey(),
        true,
        false,
        *p,
    )

    select {
    case returnNotification := <-returnChannel:
        if returnNotification.ReplyCode == amqp.NoRoute {
            return fmt.Errorf("no amqp route for %s", e.GetRoutingKey())
        }

    case <-ack:
        return nil

    case <-nack:
        return fmt.Errorf("basic nack for %s", e.GetRoutingKey())
    }

RPC

    publishing := someFunctionGeneratingAPublishing()
    publishing.ReplyTo = "amq.rabbitmq.reply-to"

    con := GetConnection()
    directChannel, _ := con.Channel()
    defer directChannel.Close()

    directChannel.Confirm(false)

    var ack = make(chan uint64)
    var nack = make(chan uint64)
    directChannel.NotifyConfirm(ack, nack)

    returnChannel := make(chan amqp.Return)
    directChannel.NotifyReturn(returnChannel)

    // consume direct-reply to pattern queue
    deliveryChan, _ := directChannel.Consume(
        "amq.rabbitmq.reply-to",
        "",
        true,
        false,
        false,
        false,
        nil,
    )


    directChannel.Publish(
        exchange,
        e.GetRoutingKey(),
        true,
        false,
        *publishing,
    )

    select {
    case returnNotification := <-returnChannel:
        if returnNotification.ReplyCode == amqp.NoRoute {
            return fmt.Errorf("no amqp route for %s", e.GetRoutingKey())
        }

    case <-ack:
        return nil

    case <-nack:
        return fmt.Errorf("basic nack for %s", e.GetRoutingKey())
    }

Answer 1

好的，在黑暗中徘徊之后，我终于意识到了我的基本错误:没有仔细阅读文档:

If the RPC server publishes with the mandatory flag set then amq.rabbitmq.reply-to.* is treated as not a queue; i.e. if the server only publishes to this name then the message will be considered "not routed"; a basic.return will be sent if the mandatory flag was set. RabbitMQ Direct-Reply-To Article

按照设计，在我配置的情况下，回复消息被视为“未路由”。