无法使用 RabbitMQ (RPC) 和 Apache Camel 进行简单的请求

Question

我希望你一切都好！首先，我是 EIP 世界的新手。我正在尝试做一个简单的请求回复：

• 一个 Golang rabbitMQ 客户端

• Kotlin 中的 apache Camel 路由充当 RabbitMQ 服务器

我试图阅读所有可能的文档并搜索答案，但我什么也没找到。我基本上是绝望的。主要是我看到了这个，但还没有任何效果。

我的目标是像图像一样进行同步请求-回复。

我的 Golang 客户端如下所示：

func (r *RabbitMQConn) GetQueue(name string) *amqp.Queue {

    ch := r.GetChannel()

    defer ch.Close()

    q, err := ch.QueueDeclare(

        name,

        false,

        false,

        true,

        false,

        nil,

    )

    if err != nil {

        panic(err)

    }

    return &q

}

func (r *RabbitMQConn) PublishAndWait(routingKey string, correlationId string, event domain.SyncEventExtSend) (domain.SyncEventExtReceive, error) {

    message, err := json.Marshal(event)

    if err != nil {

        return domain.SyncEventExtReceive{}, apperrors.ErrInternal

    }

    ch := r.GetChannel()

    defer ch.Close()

    q := r.GetQueue("response")

    h, err := ch.Consume(

        q.Name,

        "",

        true,

        false,

        false,

        false,

        nil,

    )

    if err != nil {

        return domain.SyncEventExtReceive{}, err

    }

    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)

    defer cancel()

    err = ch.PublishWithContext(

        ctx,

        "",

        routingKey,

        false,

        false,

        amqp.Publishing{

            ContentType:   "application/json",

            Body:          message,

            CorrelationId: correlationId,

            ReplyTo:       q.Name,

        },

    )

    if err != nil {

        return domain.SyncEventExtReceive{}, err

    }

    for d := range h {

        fmt.Println("Received a message:", string(d.Body))

        if d.CorrelationId == correlationId {

            var event domain.SyncEventExtReceive

            err = json.Unmarshal(d.Body, &event)

            return event, err

        }

    }

    return domain.SyncEventExtReceive{}, apperrors.ErrInternal

}

基本上，只是使用命名响应队列从默认交换中消费。此外，我将队列名称作为ReplyTo参数发送，并为其提供相关 ID。发送的路由密钥是daily-weather在这种情况下。

我问是否有人有任何简单的例子可以用 Apache Camel 做到这一点，因为我非常迷茫。如果您与我联系，可以分享任何进一步的细节。

Answer 1

解决了你好！一段时间后，我决定看一下spring-rabbitmq Camel 组件。我意识到 Camel 有exchange patterns，而 rabbitmq 默认设置为inOut. 这样，自动将信息返回给属性replyTo。  val RABBIMQ_ROUTE =       "spring-rabbitmq:default?queues={{rabbitmq.weather.daily.routing_key}}"default指的是默认交换队列。