同时运行循环中的Go语言超时

“纵有疾风来，人生不言弃”，这句话送给正在学习Golang的朋友们，也希望在阅读本文《同时运行循环中的Go语言超时》后，能够真的帮助到大家。我也会在后续的文章中，陆续更新Golang相关的技术文章，有好的建议欢迎大家在评论留言，非常感谢！

我需要在 parallel 中运行请求，而不是一个接一个地运行请求，但会超时。现在我可以用 go 来做吗？

这是我需要并行运行的具体代码，这里的技巧也是使用超时，即根据超时等待所有请求，并在完成后获取响应。

for _, test := range testers {
        checker := newtap(test.name, test.url, test.timeout)
        res, err := checker.check()
        if err != nil {
            fmt.println(err)
        }
        fmt.println(res.name)
        fmt.println(res.res.statuscode)

    }

这是所有代码（工作代码） https://play.golang.org/p/cxnjj6pw_cf

package main

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

type HT interface {
    Name() string
    Check() (*testerResponse, error)
}

type testerResponse struct {
    name string
    res  http.Response
}

type Tap struct {
    url     string
    name    string
    timeout time.Duration
    client  *http.Client
}

func NewTap(name, url string, timeout time.Duration) *Tap {
    return &Tap{
        url:    url,
        name:   name,
        client: &http.Client{Timeout: timeout},
    }
}

func (p *Tap) Check() (*testerResponse, error) {
    response := &testerResponse{}
    req, err := http.NewRequest("GET", p.url, nil)
    if err != nil {
        return nil, err
    }
    res, e := p.client.Do(req)
    response.name = p.name
    response.res = *res
    if err != nil {
        return response, e
    }
    return response, e
}

func (p *Tap) Name() string {
    return p.name
}

func main() {

    var checkers []HT

    testers := []Tap{
        {
            name:    "first call",
            url:     "http://stackoverflow.com",
            timeout: time.Second * 20,
        },
        {
            name:    "second call",
            url:     "http://www.example.com",
            timeout: time.Second * 10,
        },
    }

    for _, test := range testers {
        checker := NewTap(test.name, test.url, test.timeout)
        res, err := checker.Check()
        if err != nil {
            fmt.Println(err)
        }
        fmt.Println(res.name)
        fmt.Println(res.res.StatusCode)

        checkers = append(checkers, checker)

    }
}

解决方案

go 中流行的并发模式是使用工作池。

基本工作池使用两个通道；一个用于放置作业，另一个用于读取结果。在这种情况下，我们的作业通道将是 tap 类型，我们的结果通道将是 testerresponse 类型。

工人

从作业通道获取作业并将结果放入结果通道。

// worker defines our worker func. as long as there is a job in the
// "queue" we continue to pick up  the "next" job
func worker(jobs <-chan tap, results chan<- testerresponse) {
    for n := range jobs {
        results <- n.check()
    }
}

工作

要添加作业，我们需要迭代 testers 并将它们放入我们的作业频道。

// makejobs fills up our jobs channel
func makejobs(jobs chan<- tap, taps []tap) {
    for _, t := range taps {
        jobs <- t
    }
}

结果

为了读取结果，我们需要迭代它们。

// getresults takes a job from our worker pool and gets the result
func getresults(tr <-chan testerresponse, taps []tap) {
    for range taps {
        r := <- tr
        status := fmt.sprintf("'%s' to '%s' was fetched with status '%d'\n", r.name, r.url, r.res.statuscode)
        if r.err != nil {
            status = fmt.sprintf(r.err.error())
        }
        fmt.println(status)
    }
}

最后，我们的主要功能。

func main() {
    // make buffered channels
    buffer := len(testers)
    jobspipe := make(chan tap, buffer)               // jobs will be of type `tap`
    resultspipe := make(chan testerresponse, buffer) // results will be of type `testerresponse`

    // create worker pool
    // max workers default is 5
    // maxworkers := 5
    // for i := 0; i < maxworkers; i++ {
    //  go worker(jobspipe, resultspipe)
    // }

    // the loop above is the same as doing:
    go worker(jobspipe, resultspipe)
    go worker(jobspipe, resultspipe)
    go worker(jobspipe, resultspipe)
    go worker(jobspipe, resultspipe)
    go worker(jobspipe, resultspipe)
    // ^^ this creates 5 workers..

    makejobs(jobspipe, testers)
    getresults(resultspipe, testers)
}

把它们放在一起

我将“第二次调用”的超时更改为一毫秒，以显示超时的工作原理。

package main

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

type ht interface {
    name() string
    check() (*testerresponse, error)
}

type testerresponse struct {
    err  error
    name string
    res  http.response
    url  string
}

type tap struct {
    url     string
    name    string
    timeout time.duration
    client  *http.client
}

func newtap(name, url string, timeout time.duration) *tap {
    return &tap{
        url:    url,
        name:   name,
        client: &http.client{timeout: timeout},
    }
}

func (p *tap) check() testerresponse {
    fmt.printf("fetching %s %s \n", p.name, p.url)
    // theres really no need for newtap
    nt := newtap(p.name, p.url, p.timeout)
    res, err := nt.client.get(p.url)
    if err != nil {
        return testerresponse{err: err}
    }

    // need to close body
    res.body.close()
    return testerresponse{name: p.name, res: *res, url: p.url}
}

func (p *tap) name() string {
    return p.name
}

// makejobs fills up our jobs channel
func makejobs(jobs chan<- tap, taps []tap) {
    for _, t := range taps {
        jobs <- t
    }
}

// getresults takes a job from our jobs channel, gets the result, and
// places it on the results channel
func getresults(tr <-chan testerresponse, taps []tap) {
    for range taps {
        r := <-tr
        status := fmt.sprintf("'%s' to '%s' was fetched with status '%d'\n", r.name, r.url, r.res.statuscode)
        if r.err != nil {
            status = fmt.sprintf(r.err.error())
        }
        fmt.printf(status)
    }
}

// worker defines our worker func. as long as there is a job in the
// "queue" we continue to pick up  the "next" job
func worker(jobs <-chan tap, results chan<- testerresponse) {
    for n := range jobs {
        results <- n.check()
    }
}

var (
    testers = []tap{
        {
            name:    "1",
            url:     "http://google.com",
            timeout: time.second * 20,
        },
        {
            name:    "2",
            url:     "http://www.yahoo.com",
            timeout: time.second * 10,
        },
        {
            name:    "3",
            url:     "http://stackoverflow.com",
            timeout: time.second * 20,
        },
        {
            name:    "4",
            url:     "http://www.example.com",
            timeout: time.second * 10,
        },
        {
            name:    "5",
            url:     "http://stackoverflow.com",
            timeout: time.second * 20,
        },
        {
            name:    "6",
            url:     "http://www.example.com",
            timeout: time.second * 10,
        },
        {
            name:    "7",
            url:     "http://stackoverflow.com",
            timeout: time.second * 20,
        },
        {
            name:    "8",
            url:     "http://www.example.com",
            timeout: time.second * 10,
        },
        {
            name:    "9",
            url:     "http://stackoverflow.com",
            timeout: time.second * 20,
        },
        {
            name:    "10",
            url:     "http://www.example.com",
            timeout: time.second * 10,
        },
        {
            name:    "11",
            url:     "http://stackoverflow.com",
            timeout: time.second * 20,
        },
        {
            name:    "12",
            url:     "http://www.example.com",
            timeout: time.second * 10,
        },
        {
            name:    "13",
            url:     "http://stackoverflow.com",
            timeout: time.second * 20,
        },
        {
            name:    "14",
            url:     "http://www.example.com",
            timeout: time.second * 10,
        },
    }
)

func main() {
    // make buffered channels
    buffer := len(testers)
    jobspipe := make(chan tap, buffer)               // jobs will be of type `tap`
    resultspipe := make(chan testerresponse, buffer) // results will be of type `testerresponse`

    // create worker pool
    // max workers default is 5
    // maxworkers := 5
    // for i := 0; i < maxworkers; i++ {
    //  go worker(jobspipe, resultspipe)
    // }

    // the loop above is the same as doing:
    go worker(jobspipe, resultspipe)
    go worker(jobspipe, resultspipe)
    go worker(jobspipe, resultspipe)
    go worker(jobspipe, resultspipe)
    go worker(jobspipe, resultspipe)
    // ^^ this creates 5 workers..

    makejobs(jobspipe, testers)
    getresults(resultspipe, testers)
}

哪些输出：

// fetching http://stackoverflow.com 
// fetching http://www.example.com 
// get "http://www.example.com": context deadline exceeded (client.timeout exceeded while awaiting headers)
// 'first call' to 'http://stackoverflow.com' was fetched with status '200'

在 golang 中可以通过不同的方式实现并行性。 这是一种幼稚的方法，具有等待组、互斥体和无限的 go 例程，不推荐。 我认为使用通道是实现并行性的首选方式。

package main

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

type HT interface {
    Name() string
    Check() (*testerResponse, error)
}

type testerResponse struct {
    name string
    res  http.Response
}

type Tap struct {
    url     string
    name    string
    timeout time.Duration
    client  *http.Client
}

func NewTap(name, url string, timeout time.Duration) *Tap {
    return &Tap{
        url:  url,
        name: name,
        client: &http.Client{
            Timeout: timeout,
        },
    }
}

func (p *Tap) Check() (*testerResponse, error) {
    response := &testerResponse{}
    req, err := http.NewRequest("GET", p.url, nil)
    if err != nil {
        return nil, err
    }
    res, e := p.client.Do(req)
    if e != nil {
        return response, e
    }
    response.name = p.name
    response.res = *res

    return response, e
}

func (p *Tap) Name() string {
    return p.name
}

func main() {

    var checkers []HT
    wg := sync.WaitGroup{}
    locker := sync.Mutex{}

    testers := []Tap{
        {
            name:    "first call",
            url:     "http://google.com",
            timeout: time.Second * 20,
        },
        {
            name:    "second call",
            url:     "http://www.example.com",
            timeout: time.Millisecond * 100,
        },
    }

    for _, test := range testers {
        wg.Add(1)
        go func(tst Tap) {
            defer wg.Done()
            checker := NewTap(tst.name, tst.url, tst.timeout)
            res, err := checker.Check()
            if err != nil {
                fmt.Println(err)
            }
            fmt.Println(res.name)
            fmt.Println(res.res.StatusCode)
            locker.Lock()
            defer locker.Unlock()
            checkers = append(checkers, checker)
        }(test)
    }

    wg.Wait()
}

理论要掌握，实操不能落！以上关于《同时运行循环中的Go语言超时》的详细介绍，大家都掌握了吧！如果想要继续提升自己的能力，那么就来关注golang学习网公众号吧！