连续执行多个独立作业

学习Golang要努力，但是不要急！今天的这篇文章《连续执行多个独立作业》将会介绍到等等知识点，如果你想深入学习Golang，可以关注我！我会持续更新相关文章的，希望对大家都能有所帮助！

我有一组相互独立的工作。因此，这些作业中的每一个都可以使用 goroutine 并发运行。请注意，一旦单个作业完成，它应该等待几秒钟并重新启动（适用于所有作业），并且这会循环进行，直到 Go API 服务停止。另请注意，所有这些作业都执行相同的 goroutine（进行 REST 调用）。在 Go 中实现这一点的最佳模式是什么？请注意，我还想在服务关闭之前等待当前正在执行的作业完成。

正确答案

如果我没猜错的话，你正在寻找这样的东西 这是一个带有消费者池的服务，可以同时执行作业。当一项工作完成后，它会在一段时间后再次重复，直到您停止服务。

type job struct {
    id     int
    result chan error
}

func newjob(id int) job {
    return job{
        id:     id,
        result: make(chan error, 1),
    }
}

type service struct {
    pending chan job

    consumerlimit  int
    repeatinterval time.duration

    isclosed chan struct{}
    shutdown chan chan error
}

func newservice(repeatinterval time.duration, consumerlimit int, pendingchannelsize int) *service {
    s := &service{
        pending:        make(chan job, pendingchannelsize),
        consumerlimit:  consumerlimit,
        repeatinterval: repeatinterval,
        isclosed:       make(chan struct{}, consumerlimit),
        shutdown:       make(chan chan error),
    }

    for i := 0; i < s.consumerlimit; i++ {
        go s.consumer()
    }

    return s
}

func (s *service) do(ctx context.context, job job) error {
    select {
    case <-ctx.done():
        return ctx.err()
    case s.pending <- job:
        return <-job.result
    case <-s.isclosed:
        return errors.new("service has been shut down")
    }
}

func (s *service) consumer() {
    for {
        select {
        case j := <-s.pending:
            //simulate working process
            time.sleep(time.duration(rand.intn(200)) + 200)
            j.result <- nil
            fmt.println(fmt.sprintf("job %v is done", j.id))

            go func() {
                //repeat after a time
                time.sleep(s.repeatinterval)
                ctx, cancel := context.withtimeout(context.background(), 2*time.second)
                defer cancel()
                if err := s.do(ctx, newjob(j.id)); err != nil {
                    fmt.println(fmt.errorf("failed to send job to repeat: %v", err))
                }
            }()
        case result := <-s.shutdown:
            result <- nil
            return
        }
    }
}

func (s *service) close() error {
    result := make(chan error, 1)
    for i := 0; i < s.consumerlimit; i++ {
        s.shutdown <- result
    }
    close(s.isclosed)
    return <-result
}

func main() {
    interrupt := make(chan os.signal, 1)
    signal.notify(interrupt, os.interrupt)

    service := newservice(time.second, 5, 1000)

    //assign jobs
    for i := 1; i < 10; i++ {
        go func(i int) {
            if err := service.do(context.background(), newjob(i)); err != nil {
                fmt.println(fmt.errorf("failed to send job: %v", err))
            }
        }(i)
    }

    select {
    case <-interrupt:
        switch err := service.close(); err {
        case nil:
            fmt.println("service has been shutdown successfully")
        default:
            fmt.println(fmt.errorf("failed to graceful shut down service: %w", err))
        }
        return
    }
}

如果我理解正确的话，您正在寻找这样的东西。

此代码将在循环中运行工作人员，工作人员作为一个组并行运行，直到您退出发送结束信号的程序，但在退出之前等待当前循环完成。

func main() {
    srv := server{
        workers: 5,
    }
    srv.Run()
}

// inspired by: https://goinbigdata.com/golang-wait-for-all-goroutines-to-finish/#:~:text=A%20WaitGroup%20allows%20to%20wait,until%20all%20goroutines%20have%20finished.
func work(wg *sync.WaitGroup, i int) {
    defer wg.Done()

    rand.Seed(time.Now().UnixNano())
    n := rand.Intn(10)

    fmt.Printf("Worker %v: Started\n", i)
    time.Sleep(time.Duration(n) * time.Second)
    fmt.Printf("Worker %v: Finished\n", i)

}

type server struct {
    running bool
    workers int
}

func (srv *server) Run() {

    done := make(chan bool, 1) // this channel

    signalCh := make(chan os.Signal, 1) // this channel will get a signal on system call
    signal.Notify(signalCh, syscall.SIGINT, syscall.SIGTERM)

    go func() {
        <-signalCh
        srv.running = false
        done <- true
    }()

    srv.running = true
    for srv.running {
        var wg sync.WaitGroup
        for i := 0; i < srv.workers; i++ {
            wg.Add(1)
            go work(&wg, i)
        }
        wg.Wait()
    }
    <-done
}

今天关于《连续执行多个独立作业》的内容介绍就到此结束，如果有什么疑问或者建议，可以在golang学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！