通道传输值并读取输出的准备就绪

学习Golang要努力，但是不要急！今天的这篇文章《通道传输值并读取输出的准备就绪》将会介绍到等等知识点，如果你想深入学习Golang，可以关注我！我会持续更新相关文章的，希望对大家都能有所帮助！

我正在尝试使用 golang 中的两个通道构建接收者和发送者模式。我正在执行一项任务（api 调用），并接收回 response 结构。我的目标是，当收到响应时，我想将其发送到另一个通道（writechan）进行额外处理。

我想连续读取/监听该接收通道（respchan）并处理通过的任何内容（例如response）。然后我想启动一个线程，并在另一个 goroutine 中对该 response 进行进一步的操作。

我想了解如何将此模式链接在一起，以允许数据从 api 调用流出并同时写入（每个响应将被写入 write() 函数处理的单独文件目标。

基本上我当前的模式如下：

package main

import (
    "fmt"
    "sync"
)

func main() {

    var wg sync.WaitGroup
    respChan := make(chan Response) // Response is a struct that contains API response metadata
    defer close(respChan)
    // requests is just a slice of requests to be made to an API
    // This part is working well
    for _, req := range requests {
        wg.Add(1)
        go func(r Request) {
            defer wg.Done()
            resp, _ := r.Get() // Make the API call and receive back a Response struct
            respChan <- resp // Put the response into our channel
        }(req)
    }

    // Now, I want to extract the responses as they become available and send them to another function to do some processing. This I am unsure of how to handle properly
    writeChan := make(chan string)
    defer close(writeChan)
    select {
        case resp := <-respChan: // receive from response channel
            go func(response Response) {
                signal, _ := Write(response) // Separate func to write the response to a file. Not important here in this context.
                writeChan <- signal // Put the signal data into the channel which is a string file path of where the file was written (will be used for a later process)

            }(resp)
        case <-time.After(15 *time.Second):
            fmt.Println("15 seconds have passed without receiving anything...")

    }
    wg.Wait()
}

正确答案

让我与您分享一个您可以从中受益的工作示例。首先，我将展示代码，然后，我将引导您完成所有相关部分。

package main

import (
    "fmt"
    "net/http"
    "os"
    "strings"
    "time"
)

type request struct {
    url            string
    delayinseconds time.duration
}

type response struct {
    url        string
    statuscode int
}

func main() {
    requests := []request{
        {"https://www.google.com", 0},
        {"https://stackoverflow.com", 1},
        {"https://www.wikipedia.com", 4},
    }

    respchan := make(chan response)
    defer close(respchan)

    for _, req := range requests {
        go func(r request) {
            fmt.printf("%q - %v\n", r.url, strings.repeat("#", 30))
            // simulate heavy work
            time.sleep(time.second * r.delayinseconds)
            resp, _ := http.get(r.url)
            res := response{r.url, resp.statuscode}
            fmt.println(time.now())
            respchan <- res
        }(req)
    }

    writechan := make(chan struct{})
    defer close(writechan)

    for i := 0; i < len(requests); i++ {
        select {
        case res := <-respchan:
            go func(r response) {
                f, err := os.create(fmt.sprintf("%v.txt", strings.replace(r.url, "https://", "", 1)))
                if err != nil {
                    panic(err)
                }
                defer f.close()
                f.write([]byte(fmt.sprintf("%q ok with %d\n", r.url, r.statuscode)))
                writechan <- struct{}{}
            }(res)
        case <-time.after(time.second * 2):
            fmt.println("timeout")
        }
    }
}

设置

首先，我定义了示例中将使用的两个结构：request 和 response。在前者中，我放置了一个 delayinseconds 来模拟一些重负载和耗时的操作。然后，我定义了 requests 变量，其中包含必须完成的所有请求。

写作部分

在这里，我检查了 requests 变量。对于每个请求，我将向目标 url 发出 http 请求。 time.sleep 模拟重负载。然后，我将响应写入未缓冲的 respchan 通道。

阅读部分

这里的主要变化是将 select 构造包装到 for 循环中。因此，我们将确保迭代正确的次数（基于 requests 变量的长度）。

最后的注释

首先，请记住，代码过于简单只是为了展示相关部分。因此，缺少许多错误处理，并且一些内联函数可以外推到命名函数中。你不需要使用sync.waitgroup来实现你所需要的，通道的使用就足够了。
随意尝试延迟并检查写入了哪些文件！

如果这对您有帮助，请告诉我！

编辑

根据您的要求，我将根据您的需求为您提供更准确的解决方案。新的阅读部分将类似于以下内容：

count := 0
for {
    // this check is need to exit the for loop and not wait indefinitely
    // it can be removed based on your needs
    if count == 3 {
        fmt.Println("all responses arrived...")
        return
    }
    res := <-respChan
    count++
    go func(r Response) {
        f, err := os.Create(fmt.Sprintf("%v.txt", strings.Replace(r.Url, "https://", "", 1)))
        if err != nil {
            panic(err)
        }
        defer f.Close()
        f.Write([]byte(fmt.Sprintf("%q OK with %d\n", r.Url, r.StatusCode)))
        writeChan <- struct{}{}
    }(res)
}

这里，执行在 for 循环内无限期等待。无论每个请求需要多长时间才能完成，它一到达就会被获取。我在 for 循环的顶部放置了一个 if ，以便在处理完我们需要的请求后退出。但是，您可以避免它并让代码运行直到收到取消信号（这取决于您）。

请告诉我这是否更能满足您的要求，谢谢！

以上就是本文的全部内容了，是否有顺利帮助你解决问题？若是能给你带来学习上的帮助，请大家多多支持golang学习网！更多关于Golang的相关知识，也可关注golang学习网公众号。