超时后关闭通道

Question

我有一个小型 Go 程序，它每滴答声（1 秒）发出许多请求。我正在尝试同时提出这些请求。我想计算并记录一次成功请求的数量，然后继续。如果请求没有及时完成，我不想阻止主代码。

下面的代码实现了这一点，但我不相信我在 concurrentReqs 中正确关闭了通道。因为任何错过截止日期的请求仍会使用前一个刻度记录。我也相信 main 函数中的代码会阻塞等待 concurrentReqs 完成。我尝试在我的选择中将close(ch) 移动到超时情况下，但这会导致“在关闭通道上发送”错误。

我的理解是，使用具有截止日期的上下文（可能在我的主要股票代码中设置）可能是解决此问题的方法，但我正在努力解决这些问题，我想知道是否还有其他可以尝试的方法。

注意：concurrentReqs 中的超时时间故意设置得很低，因为我在本地进行测试。

package main

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

type response struct {
    num int
    statusCode int
    requestDuration time.Duration
}

func singleRequest(url string, i int, tick int) response {
    start := time.Now()
    client := http.Client{ Timeout: 100 * time.Millisecond }

    resp, _ := client.Get(url)
    fmt.Printf("%d: %d\n", tick, i)

    defer resp.Body.Close()

    return response{statusCode: int(resp.StatusCode), requestDuration: time.Since(start)}
}

func concurrentReqs(url string, reqsPerTick int, tick int) (results []response){
    ch := make(chan response, reqsPerTick)
    timeout := time.After(20 * time.Millisecond) // deliberately low
    results = make([]response, 0)

    for i := 0; i < reqsPerTick; i++ {
        go func(i int, t int) {
            ch <- singleRequest(url, i, tick)
        }(i, tick)
    }

    for i := 0; i < reqsPerTick; i++ {
        select {
        case response := <- ch:
            results = append(results, response)
        case <- timeout:
            return
        }
    }
    close(ch)

    return
}

func main() {
    var url string = "http://end-point.svc/req"

    c := time.Tick(1 * time.Second)
    for next := range c {
        things := concurrentReqs(url, 100, next.Second())
        fmt.Printf("%s: Successful Reqs - %d\n", int(next.Second()), len(things))
    }
}

Answer 1

我建议使用带超时的上下文来取消和超时。此外，我认为使用等待组和互斥锁保护的结果写入有助于消除第二个循环。

package main

import (
    "context"
    "fmt"
    "log"
    "net/http"
    "sync"
    "time"
)

type response struct {
    num             int
    statusCode      int
    requestDuration time.Duration
}

func singleRequest(ctx context.Context, url string, i int, tick int) (response, error) {
    start := time.Now()

    req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
    if err != nil {
        return response{requestDuration: time.Since(start)}, err
    }

    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        return response{requestDuration: time.Since(start)}, err
    }

    fmt.Printf("%d: %d\n", tick, i)

    defer resp.Body.Close()

    return response{statusCode: int(resp.StatusCode), requestDuration: time.Since(start)}, nil
}

func concurrentReqs(url string, reqsPerTick int, tick int) (results []response) {
    mu := sync.Mutex{}
    results = make([]response, 0)

    ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
    defer cancel()

    wg := sync.WaitGroup{}
    for i := 0; i < reqsPerTick; i++ {
        wg.Add(1)
        go func(i int, t int) {
            defer wg.Done()
            response, err := singleRequest(ctx, url, i, tick)
            if err != nil {
                log.Print(err)
                return
            }
            mu.Lock()
            results = append(results, response)
            mu.Unlock()
        }(i, tick)
    }

    wg.Wait()

    return results
}

func main() {
    var url string = "http://end-point.svc/req"

    c := time.Tick(1 * time.Second)
    for next := range c {
        // You may want to wrap this in a goroutine to make sure tick is not skipped.
        // Otherwise if concurrentReqs takes more than a tick time for whatever reason, a tick will be skipped.
        things := concurrentReqs(url, 100, next.Second())
        fmt.Printf("%s: Successful Reqs - %d\n", int(next.Second()), len(things))
    }
}