go - 如何处理同一个 RabbitMQ 队列中的多个 protobuf 消息？

Question

我的问题是我使用单个队列(作为我的服务的入口点)并使用 Go 消费者来处理传入的消息。

我的消费者

    message := pb.GetRequest{}
    err := proto.Unmarshal(msg.Body, message)

我的问题是我的消费者只能处理 GetRequests。如果我需要处理其他类型的消息，即。 AddRequest 要么

我需要为每条消息或

定义一个新队列

我需要先看看是否可以先unmartial(GetRequest)，然后继续测试是否可以unmartialed到(AddRequest)

有没有其他好的方法可以做到这一点(假设 #1 不是一个好的选择)

Answer 1

使用 switch在 RabbitMQ 路由键上。Channel.Consume方法返回 <-chan amqp.Delivery 类型的 Go channel , 其中 amqp.Delivery 包含字段 RoutingKey .
路由键是用于将发布的消息与消费者订阅匹配的标识符。您应该确保您的发布者在路由键和消息类型之间保持一对一的关联。
发布者代码将如下所示:

msg := &pb.AddRequest{} // some protobuf generated type
body, _ := proto.Marshal(msg)
err := ch.Publish(
        "my-exchange", // exchange name
        "foo.bar.add", // routing key
        true,          // option: mandatory
        true,          // option: immediate
        amqp.Publishing{
            ContentType: "application/x-protobuf",
            Body:        body,
        },
    )

在上面的示例中，您必须确保 全部且仅 *pb.AddRequest 类型的消息使用路由键 foo.bar.add 发布，即您的消息类型是确定性的。
如果你能做到这一点，那么你的消费者代码可以打开路由键并将 MQ 有效负载解码为正确类型的变量:

func formatEvent(payload amqp.Delivery) (proto.Message, error) {

    var event proto.Message

    // switch on the routing key
    switch payload.RoutingKey {
    case "foo.bar.add":
        event = &pb.AddRequest{}
    case "foo.bar.get":
        event = &pb.GetRequest{}
    default:
        return nil, fmt.Errorf("unknown routingKey: %s", key)
    }

    // unmarshal the body into the event variable
    if err := proto.Unmarshal(payload.Body, event); err != nil {
        return nil, err
    }

    return event, nil
}

然后你可以 type-switch在 proto.Message实例来处理每个具体的消息类型。 (当然您也可以直接在路由键开关中处理具体消息；这将取决于您希望如何组织代码)。