go - 让 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 之前关闭.