工作池，带有固定轮询时间和缓冲作业

编程并不是一个机械性的工作，而是需要有思考，有创新的工作，语法是固定的，但解决问题的思路则是依靠人的思维，这就需要我们坚持学习和更新自己的知识。今天golang学习网就整理分享《工作池，带有固定轮询时间和缓冲作业》，文章讲解的知识点主要包括，如果你对Golang方面的知识点感兴趣，就不要错过golang学习网，在这可以对大家的知识积累有所帮助，助力开发能力的提升。

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

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

请参阅下面的示例代码，在此处的演示中运行它。

是否可以在没有原子计数器的情况下重写此代码来跟踪正在进行的作业？

// 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")
}

解决方案

这是代码的基于通道的版本，其功能与上面示例的意图相同。关键点是我们没有使用任何原子值来改变代码的逻辑，因为这不提供 goroutine 之间的同步。 goroutine 之间的所有交互都使用通道 sync.waitgroup 或 context.context 进行同步。可能有更好的方法来解决手头的问题，但这表明不需要原子来协调队列和工作人员。

这里 goroutine 之间唯一未协调的值是在日志输出中使用 len(jobs)。使用它是否有意义取决于您，因为它的值在并发世界中毫无意义，但它是安全的，因为它是同步使用的，并且没有基于该值的逻辑。

buf := 5
workers := 3
jobs := make(chan int, buf)

// results buffer must always be larger than workers + buf to prevent deadlock
results := make(chan int, buf*2)

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

// Start workers
var wg sync.WaitGroup
for n := 0; n < workers; n++ {
    wg.Add(1)
    go func(n int) {
        defer wg.Done()
        for jobID := range jobs {
            fmt.Printf("worker %v processing %v - %v jobs left\n", n, jobID, len(jobs))
            time.Sleep(time.Duration(rand.Intn(5)) * pollInterval)
            results <- jobID
        }
        fmt.Printf("worker %v exited", n)
    }(n)
}

var done sync.WaitGroup
done.Add(1)
go func() {
    defer done.Done()
    ticker := time.NewTicker(pollInterval)
    r := make([]string, 0)

    flushResults := func() {
        fmt.Printf("===> results: %v\n", strings.Join(r, ","))
        r = r[:0]
    }

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

            // send max buf jobs, or fill the queue
            for i := 0; i < buf; i++ {
                jobID++
                select {
                case jobs <- jobID:
                    continue
                }
                break
            }
            fmt.Printf("===> send %v jobs\n", i)

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

        case <-ctx.Done():
            // Close jobs channel to stop workers
            close(jobs)
            // Wait for workers to exit
            wg.Wait()

            // we can close results for easy iteration because we know
            // there are no more workers.
            close(results)
            // Flush remaining results
            for jobID := range results {
                r = append(r, fmt.Sprintf("%v", jobID))
            }
            flushResults()
            return
        }
    }
}()

今天关于《工作池，带有固定轮询时间和缓冲作业》的内容介绍就到此结束，如果有什么疑问或者建议，可以在golang学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！