具有缓冲作业和固定轮询间隔的工作池

Question

我有一个工作池在工作频道上侦听，并在结果信道上进行响应。

作业生产者必须以固定的行情间隔运行。在读取足够多的新作业以填充缓冲区之前，必须先刷新结果。 批处理刷新结果并批量读取新作业至关重要。>

请参见下面的示例代码，在playground here上运行。

是否可以在没有原子计数器的情况下重写此记录以跟踪inflight

作业？

// Worker pool with buffered jobs and fixed polling interval

package main

import (
    "fmt"
    "math/rand"
    "os"
    "os/signal"
    "strings"
    "sync"
    "sync/atomic"
    "syscall"
    "time"
)

func main() {
    rand.Seed(time.Now().UnixNano())

    // buf is the size of the jobs buffer
    buf := 5

    // workers is the number of workers to start
    workers := 3

    // jobs chan for workers
    jobs := make(chan int, buf)
    // results chan for workers
    results := make(chan int, buf*2)

    // jobID is incremented for each job sent on the jobs chan
    var jobID int

    // inflight is a count of the items in the jobs chan buffer
    var inflight uint64

    // pollInterval for jobs producer
    pollInterval := 500 * time.Millisecond

    // pollDone chan to stop polling
    pollDone := make(chan bool)

    // jobMultiplier on pollInterval for random job processing times
    jobMultiplier := 5

    // done chan to exit program
    done := make(chan bool)

    // Start workers
    wg := sync.WaitGroup{}
    for n := 0; n < workers; n++ {
        wg.Add(1)
        go (func(n int) {
            defer wg.Done()
            for {
                // Receive from channel or block
                jobID, more := <-jobs
                if more {
                    // To subtract a signed positive constant value...
                    // https://golang.org/pkg/sync/atomic/#AddUint64
                    c := atomic.AddUint64(&inflight, ^uint64(0))
                    fmt.Println(
                        fmt.Sprintf("worker %v processing %v - %v jobs left",
                            n, jobID, c))
                    // Processing the job...
                    m := rand.Intn(jobMultiplier)
                    time.Sleep(time.Duration(m) * pollInterval)
                    results <- jobID
                } else {
                    fmt.Println(fmt.Sprintf("worker %v exited", n))
                    return
                }
            }
        })(n)
    }

    // Signal to exit
    sig := make(chan os.Signal, 1)
    signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
    fmt.Println("ctrl+c to exit")

    go (func() {
        ticker := time.NewTicker(pollInterval)
        r := make([]string, 0)
        flushResults := func() {
            fmt.Println(
                fmt.Sprintf("===> results: %v", strings.Join(r, ",")))
            r = make([]string, 0)
        }

        for {
            select {
            case <-ticker.C:
                flushResults()

                // Fetch jobs
                c := atomic.LoadUint64(&inflight)
                d := uint64(buf) - c
                for i := 0; i < int(d); i++ {
                    jobID++
                    jobs <- jobID
                    atomic.AddUint64(&inflight, 1)
                }
                fmt.Println(fmt.Sprintf("===> send %v jobs", d))

            case jobID := <-results:
                r = append(r, fmt.Sprintf("%v", jobID))

            case <-pollDone:
                // Stop polling for new jobs
                ticker.Stop()

                // Close jobs channel to stop workers
                close(jobs)

                // Wait for workers to exit
                wg.Wait()
                close(results)

                // Flush remaining results
                for {
                    jobID, more := <-results
                    if more {
                        r = append(r, fmt.Sprintf("%v", jobID))
                    } else {
                        break
                    }
                }
                flushResults()

                // Done!
                done <- true
                return
            }
        }
    })()

    // Wait for exit signal
    <-sig

    fmt.Println("---------| EXIT |---------")
    pollDone <- true
    <-done
    fmt.Println("...done")
}

我有一个工作人员池在工作频道上侦听，并在结果频道上进行响应。作业生产者必须以固定的行情间隔运行。必须刷新结果，然后才能读取足够多的新内容...

Answer 1

这里是您的代码的基于通道的版本，在功能上等同于以上示例的意图。关键是我们没有使用任何原子值来改变代码的逻辑，因为这在goroutine之间没有提供同步。使用通道sync.WaitGroup或context.Context同步goroutine之间的所有交互。解决当前问题可能有更好的方法，但这表明不需要协调队列和工作程序的原子。