让 go 例程等待 rabbitMQ 发送结果

Question

我是 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")
}

rabbitReceive方法：

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 。提前致谢:)

Answer 1

修改您的 func rabbitReceive(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 之前关闭。

Answer 2

跟进 @nightfury1204 的好答案，您确实在写信给 resCh 之前关闭了 ch。只有一件事，在 go 例程中你想要遍历 all 消息，所以更好的方法是：

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)
    }()

让 go 例程等待 rabbitMQ 发送结果

Make go routine wait until result from rabbitMQ is sent

synchronization

go

rabbitmq

goroutine