大家好，我们又见面了啊~本文《使用带有时间和超时的作业》的内容中将会涉及到等等。如果你正在学习Golang相关知识，欢迎关注我，以后会给大家带来更多Golang相关文章，希望我们能一起进步！下面就开始本文的正式内容~

我使用以下有效的示例代码，现在我希望每个作业都能够 打印执行所花费的时间（最好是通用的，不是每个作业都需要使用代码

start := time.now()
took := time.since(start).milliseconds()

并且还为作业提供超时，例如，如果需要超过 10 秒才能终止或停止它。

package main

import (
    "encoding/json"
    "fmt"

    "github.com/gammazero/workerpool"
)

var numworkers = 10

type myreturntype struct {
    name string
    data interface{}
}

func wrapjob(rc chan myreturntype, f func() myreturntype) func() {
    return func() {
        rc <- f()
    }
}

func main() {
    // create results chan and worker pool
    // should prob make your results channel typed to what you need
    jobs := []func() myreturntype {
        func() myreturntype {
           return job1()
        },
        func() myreturntype {
           return job2()
        },
    }

    results := make(chan myreturntype, len(jobs))
    pool := workerpool.new(numworkers)

    for _, job := range jobs {
        j := job
        pool.submit(wrapjob(results, j))
    }

    // wait for all jobs to finish
    pool.stopwait()

    // close results chan
    close(results)

    // iterate over results, printing to console
    for res := range results {
        prettyprint(res)
    }
}

func prettyprint(i interface{}) {
    prettyjson, err := json.marshalindent(i, "", "    ")
    if err != nil {
        fmt.printf("error : %s \n", err.error())
    }
    fmt.printf("myreturntype %s\n", string(prettyjson))
}

以下是我试图避免的示例，并为每个作业的打印时间提供一些通用解决方案：

func job1() {
   start := time.Now()
   ...
   // running some code
   took := time.Since(start).Milliseconds()
}

func job2(){
   start := time.Now()
   ...
   // running some code
   took := time.Since(start).Milliseconds()
}

解决方案

更新

滚动到“这是已接受的答案”部分以查看已接受的答案

我继续根据接受的答案中的代码编写了一个小库......

You can find it here 或以下代码：

// how to use the library
package main

import (
    "fmt"
    "time"

    "github.com/oze4/reactor"
)

func main() {
    timeoutforjobs := time.duration(time.second * 10)
    numofworkers := 10

    myreactor := reactor.new(numofworkers, timeoutforjobs)

    // you can also create a reactor with a custom client
    // myreactor := reactor.newwithclient(numofworkers, timeoutforjobs, &reactor.client{...})

    // add job(s)
    myreactor.add(reactor.job{
        name: "job1",
        runner: func(c *reactor.client) reactor.react {
            // do something with client `c`
            res, _ := c.http.get("xyz.com")
            return reactor.react{info: res}
        },
    })

    // all results will be here
    results := myreactor.getresults()
    
    for _, result := range results {
        fmt.println(result)
    }
}

库代码

// library code
package reactor

import (
    "context"
    "net/http"
    "time"

    "github.com/gammazero/workerpool"
    "k8s.io/client-go/kubernetes"
)

// new creates a new reactor
func new(maxworkers int, jobtimeout time.duration) reactor {
    // do whatever you need to here to create default client
    defaultclient := &client{
        http:       http.client{},
        kubernetes: kubernetes.clientset{},
    }

    return &reactor{
        workerpool:  workerpool.new(maxworkers),
        jobtimeout:  jobtimeout,
        transport:   defaultclient,
        resultschan: make(chan react, 100),
    }
}

// newwithclient creates a new reactor with a custom client
func newwithclient(client *client, maxworkers int, jobtimeout time.duration) reactor {
    return &reactor{
        workerpool:  workerpool.new(maxworkers),
        jobtimeout:  jobtimeout,
        transport:   client,
        resultschan: make(chan react, 100),
    }
}

// reactor knows how to handle jobs
type reactor interface {
    add(j job)                          // add puts a job on the queue
    client() *client                    // i dont know if you want the consumer to have access to this or not
    getresults() []react                // get results
    timeout() time.duration             // i dont know if you want the consumer to have access to this or not
    workerpool() *workerpool.workerpool // i dont know if you want the consumer to have access to this or not
}

type reactor struct {
    jobtimeout  time.duration
    workerpool  *workerpool.workerpool
    resultschan chan react
    transport   *client
}

// add submits a job
func (r *reactor) add(j job) {
    r.workerpool.submit(r.wrapper(j))
}

// i dont know if you want the consumer to have access to this or not
func (r *reactor) client() *client {
    return r.transport
}

// get results gets results
func (r *reactor) getresults() []react {
    return r.getresults()
}

func (r *reactor) getresults() []react {
    r.workerpool.stopwait()
    close(r.resultschan)

    var allreacts []react
    for jobreact := range r.resultschan {
        allreacts = append(allreacts, jobreact)
    }

    return allreacts
}

func (r *reactor) timeout() time.duration {
    return r.jobtimeout
}

// i dont know if you want the consumer to have access to this or not
func (r *reactor) workerpool() *workerpool.workerpool {
    return r.workerpool
}

// worker should be private
func (r *reactor) worker(ctx context.context, done context.cancelfunc, job job, start time.time) {
    runner := job.runner(r.transport)
    runner.duration = time.since(start)
    runner.name = job.name

    if ctx.err() == nil {
        r.resultschan <- runner
    }

    done()
}

// wrapper should be private
func (r *reactor) wrapper(job job) func() {
    ctx, cancel := context.withtimeout(context.background(), r.jobtimeout)

    return func() {
        start := time.now()
        go r.worker(ctx, cancel, job, start)

        select {
        case <-ctx.done():
            switch ctx.err() {
            case context.deadlineexceeded:
                r.resultschan <- react{
                    error:    context.deadlineexceeded,
                    name:     job.name,
                    duration: time.since(start),
                }
            }
        }
    }
}

// react holds response data
type react struct {
    // this should be public so the consumer can set it
    info  interface{}
    error error

    // these fields should be private and handled via public methods
    duration time.duration
    name     string
}

// duration returns duration
func (r *react) duration() time.duration {
    return r.duration
}

// name returns the job name
func (r *react) name() string {
    return r.name
}

// client holds http and kubernetes clients
type client struct {
    http       http.client
    kubernetes kubernetes.clientset
}

// job holds job data
type job struct {
    name   string
    runner func(*client) react
}

这是已接受的答案

以下示例展示了如何收集执行时间以及设置超时..

package main

import (
    "context"
    "fmt"
    "time"

    "github.com/gammazero/workerpool"
)

var (
    //
    // set timeout for all jobs here
    //
    jobtimeout = time.duration(time.second * 1)
)

// myreturntype could be anything you want it to be
type myreturntype struct {
    name              string
    data              interface{}
    error             error
    executionduration time.duration
}

// name returns name. it is written like this so the consumer
// cannot change the name outside of supplying one via the job
func (m *myreturntype) name() string {
    return m.name
}

// job holds job data
type job struct {
    name string
    task func() myreturntype
}

func wrapjob(timeout time.duration, resultschan chan myreturntype, job job) func() {
    timeoutcontext, timeoutcancel := context.withtimeout(context.background(), timeout)

    return func() {
        timerstart := time.now()
        go func(ctx context.context, done context.cancelfunc, reschan chan myreturntype, todo job, starttime time.time) {
            result := todo.task()
            result.executionduration = time.since(starttime)
            result.name = todo.name
            if timeoutcontext.err() == nil {
                reschan <- result
            }
            done()
        }(timeoutcontext, timeoutcancel, resultschan, job, timerstart)

        select {
        case <-timeoutcontext.done():
            switch timeoutcontext.err() {
            case context.deadlineexceeded:
                resultschan <- myreturntype{
                    name:              job.name,
                    error:             context.deadlineexceeded,
                    executionduration: time.since(timerstart),
                }
            }
        }
    }
}

func main() {
    jobs := []job{
        {
            name: "job1",
            task: func() myreturntype {
                // this will surpass our timeout and should get cancelled
                time.sleep(time.second * 3)
                // don't have to set the name here
                return myreturntype{data: map[string]string{"whatever": "you want"}}
            },
        },
        {
            name: "job2",
            task: func() myreturntype {
                // this job will succeed
                time.sleep(time.millisecond * 300)
                resultfromcurl := "i am a result"
                return myreturntype{data: resultfromcurl}
            },
        },
    }

    jobresultschannel := make(chan myreturntype, len(jobs))
    pool := workerpool.new(10)

    for _, job := range jobs {
        pool.submit(wrapjob(jobtimeout, jobresultschannel, job))
    }

    pool.stopwait()
    close(jobresultschannel)

    // do whatever you want with results
    for jobresult := range jobresultschannel {
        if jobresult.error != nil {
            fmt.printf("[took '%d' ms] '%s' : joberror : %s\n", jobresult.executionduration, jobresult.name(), jobresult.error)
        } else {
            fmt.printf("[took '%d' ms] '%s' : jobsuccess : %s\n", jobresult.executionduration, jobresult.name(), jobresult.data)
        }
    }
}

返回结果：

// [took '305182398' ms] 'job2' : jobsuccess : i am a result
// [took '1001045539' ms] 'job1' : joberror : context deadline exceeded

原始答案

您应该能够使用上下文进行超时/取消 (as Peter mentioned)。

就记录执行时间而言，您可以执行 what you stated in your comment 或类似的操作：

package main

import (
    "fmt"
    "time"

    "github.com/gammazero/workerpool"
)

type MyReturnType struct {
    Name string
    Data interface{}
    Time time.Duration
}

func wrapJob(rc chan MyReturnType, f func() MyReturnType) func() {
    return func() {
        start := time.Now()
        result := f()
        result.Time = time.Since(start)
        rc <- result
    }
}

func main() {
    jobs := []func() MyReturnType{
        func() MyReturnType {
            time.Sleep(time.Millisecond*400)
            return MyReturnType{Name: "job1", Data: map[string]string{"Whatever": "You want"}}
        },
        func() MyReturnType {
            resultFromCurl := "i am a result"
            return MyReturnType{Name: "job2", Data: resultFromCurl}
        },
    }

    results := make(chan MyReturnType, len(jobs))
    pool := workerpool.New(10)

    for _, job := range jobs {
        j := job
        pool.Submit(wrapJob(results, j))
    }

    pool.StopWait()
    close(results)

    for res := range results {
        fmt.Printf("[took '%d' ms] ", res.Time)
        fmt.Println(res)
    }
}

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