进行例行检查，直到发送出RabbitMQ的结果

Question

I am fairly new to Go, I want to make a pipeline that translate every requests I receive by send it to first queue (TEST), and get the final result from the last queue (RESULT) and send it back as a response.

The problem I am facing is, the response never wait til all result back from the queue. Here is the code:

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 method:

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 method:

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
}

Here is what I get when I run the program:

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]

What I want is that, I receive the result exactly after I send the request, so the request will not get the wrong result as a response. Something like:

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]

I believe I did not use goroutine or sync.WaitGroup correctly here. Thanks in advance :)

Answer 1

Modify your func rabbitReceive(queueName string) <-chan string as below:

 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
}

The reason previous code cause you problem was defer ch.Close(). ch closes before response was written to resCh.