与超时并行运行循环

问题描述

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

这是我需要在 parallel 中运行的特定代码，这里的窍门也是使用超时，即根据超时等待所有请求并在完成后获取响应。

    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 _,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类型。

工人

从Jobs频道获取一份工作，然后将结果放入结果频道。

// 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)
    // ^^ 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,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.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",{
            name:    "4",url:     "http://www.example.com",{
            name:    "5",{
            name:    "6",{
            name:    "7",{
            name:    "8",{
            name:    "9",{
            name:    "10",{
            name:    "11",{
            name:    "12",{
            name:    "13",{
            name:    "14",}
)

func main() {
    // Make buffered channels
    buffer := len(testers)
    jobsPipe := make(chan Tap,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中以不同的方式完成。 不建议使用等待组，Mutex和无限制执行例程的幼稚方法。 我认为使用通道是进行并行处理的首选方法。

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,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",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",{
            name:    "second call",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()
}