等待rabbitMQ结果后启动go例程

RabbitMQ 是一种流行的消息代理，用于在分布式系统中进行异步消息传递。它提供可靠的消息传递、队列和路由等功能。本文将指导您使用 Go 语言与 RabbitMQ 集成，包括发送和接收消息、声明队列和通道等基本操作。通过遵循本文中的步骤，您可以轻松地将 RabbitMQ 集成到您的 Go 应用程序中，并利用其强大功能来提高应用程序的可靠性和可扩展性。

我对 go 还很陌生，我想创建一个管道，通过将其发送到第一个队列（test）来转换我收到的每个请求，并从最后一个队列（result）获取最终结果并将其作为回应。

我面临的问题是，响应永远不会等到所有结果从队列返回。这是代码：

func main() {
    requests := []int{3, 4, 5, 6, 7}
    var wg sync.waitgroup
    wg.add(1)
    resarr := []string{}
    go func() {
        for _, r := range requests {
            rabbitsend("test", r)
            resarr = append(resarr, <-rabbitreceive("result"))
        }
        defer wg.done()
    }()
    wg.wait()

    log.println("result", resarr)
}

rabbitsend方法：

func rabbitsend(queuename string, msg int) {
    conn, err := amqp.dial("amqp://guest:guest@localhost:5672/")
    failonerror(err, "failed to connect to rabbitmq")
    defer conn.close()

    ch, err := conn.channel()
    failonerror(err, "failed to open a channel")
    defer ch.close()

    q, err := ch.queuedeclare(
        queuename, // name
        true,      // durable
        false,     // delete when unused
        false,     // exclusive
        false,     // no-wait
        nil,       // arguments
    )
    failonerror(err, "failed to declare a queue")

    body, _ := json.marshal(msg)
    err = ch.publish(
        "",     // exchange
        q.name, // routing key
        false,  // mandatory
        false,  // immediate
        amqp.publishing{
            contenttype: "application/json",
            body:        []byte(body),
        })
    log.printf("[x] sent %s to %s", body, q.name)
    failonerror(err, "failed to publish a message")
}

rabbit接收方法：

func rabbitreceive(queuename string) <-chan string {
    conn, err := amqp.dial("amqp://guest:guest@localhost:5672/")
    failonerror(err, "failed to connect to rabbitmq")
    defer conn.close()

    ch, err := conn.channel()
    failonerror(err, "failed to open a channel")
    defer ch.close()

    q, err := ch.queuedeclare(
        queuename, // name
        true,      // durable
        false,     // delete when usused
        false,     // exclusive
        false,     // no-waits
        nil,       // arguments
    )
    failonerror(err, "failed to declare a queue")

    msgs, err := ch.consume(
        q.name, // queue
        "",     // consumer
        true,   // auto-ack
        false,  // exclusive
        false,  // no-local
        false,  // no-wait
        nil,    // args
    )
    failonerror(err, "failed to register a consumer")

    resch := make(chan string)
    go func() {
        for d := range msgs {
            log.printf("received a message: %v from %v", string(d.body), q.name)
            resch <- string(d.body)
        }
        close(resch)
    }()
    return resch
}

这是我运行程序时得到的结果：

2018/11/12 05:11:54 [x] sent 3 to test
2018/11/12 05:11:54 [x] sent 4 to test
2018/11/12 05:11:54 received a message: 9 from result
2018/11/12 05:11:54 [x] sent 5 to test
2018/11/12 05:11:54 [x] sent 6 to test
2018/11/12 05:11:54 received a message: 15 from result
2018/11/12 05:11:54 [x] sent 7 to test
2018/11/12 05:11:54 received a message: 18 from result
2018/11/12 05:11:54 result [ 9  15 18]

我想要的是，我在发送请求后立即收到结果，这样请求就不会得到错误的结果作为响应。像这样的东西：

2018/11/12 05:11:54 [x] Sent 3 to TEST
2018/11/12 05:11:54 Received a message: 9 from RESULT
2018/11/12 05:11:54 [x] Sent 4 to TEST
2018/11/12 05:11:54 Received a message: 12 from RESULT
2018/11/12 05:11:54 [x] Sent 5 to TEST
2018/11/12 05:11:54 Received a message: 15 from RESULT
2018/11/12 05:11:54 [x] Sent 6 to TEST
2018/11/12 05:11:54 Received a message: 18 from RESULT
2018/11/12 05:11:54 [x] Sent 7 to TEST
2018/11/12 05:11:54 Received a message: 21 from RESULT
2018/11/12 05:11:54 Result [ 9 12 15 18 21]

我相信我在这里没有正确使用goroutine或sync.waitgroup。预先感谢:)

解决方案

修改您的 funcrabbitreceive(queuename string) <-chan string 如下：

func rabbitreceive(queuename string) <-chan string {
    conn, err := amqp.dial("amqp://guest:guest@localhost:5672/")
    failonerror(err, "failed to connect to rabbitmq")

    ch, err := conn.channel()
    failonerror(err, "failed to open a channel")

    q, err := ch.queuedeclare(
        queuename, // name
        true,      // durable
        false,     // delete when usused
        false,     // exclusive
        false,     // no-waits
        nil,       // arguments
    )
    failonerror(err, "failed to declare a queue")

    msgs, err := ch.consume(
        q.name, // queue
        "",     // consumer
        true,   // auto-ack
        false,  // exclusive
        false,  // no-local
        false,  // no-wait
        nil,    // args
    )
    failonerror(err, "failed to register a consumer")

    resch := make(chan string)
    go func() {
        d := <-msgs
        log.printf("received a message: %v from %v", string(d.body), q.name)
        resch <- string(d.body)
        conn.close()
        ch.close()
        close(resch)
    }()
    return resch
}

之前的代码导致您出现问题的原因是 defer ch.close()。 ch 在响应写入 resch 之前关闭。

跟进 @nightfury1204 很好的答案，您确实在写入 resch 之前关闭 ch。只是一件事，在 go 例程中，您想要查看所有消息，因此更好的方法是：

go func() {
        for d := range msgs {
          log.Printf("Received a message: %v from %v", string(d.Body), q.Name)
          resCh <- string(d.Body)  
        }
        conn.Close()
        ch.Close()
        close(resCh)
    }()

理论要掌握，实操不能落！以上关于《等待rabbitMQ结果后启动go例程》的详细介绍，大家都掌握了吧！如果想要继续提升自己的能力，那么就来关注golang学习网公众号吧！