Заставьте рутинное ожидание, пока результат от rabbitMQ не будет отправлен - PullRequest
Новая
       20

Заставьте рутинное ожидание, пока результат от rabbitMQ не будет отправлен

0 голосов
/ 12 ноября 2018

Я довольно новичок в 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")
}

Способ кроликаПолучить: 

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 . Заранее спасибо:)

Ответы [ 2 ]

0 голосов
/ 12 ноября 2018

В ответ на @ nightfury1204 отличный ответ, вы действительно закрываете ch перед тем, как писать resCh. только одна вещь, в процедуре 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)
    }()
0 голосов
/ 12 ноября 2018

Измените ваш 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.

...