Go 常见并发模式实现（一）：调度后台处理任务的作业程序

package runner

import (
    "errors"
    "os"
    "os/signal"
    "time"
)

type JobRunner struct {
    interrupt chan os.Signal
    complete chan error
    timeout <-chan time.Time
    tasks []func(int)
}

var ErrTimeout = errors.New("received timeout")
var ErrInterrupt = errors.New("received interrupt")

func New(d time.Duration) *JobRunner {
    return &JobRunner{
        interrupt: make(chan os.Signal, 1),
        complete:  make(chan error),
        timeout:   time.After(d),
    }
}

func (r *JobRunner) Add(tasks ...func(int)) {
    r.tasks = append(r.tasks, tasks...)
}

func (r *JobRunner) Start() error {
    // 接收系统中断信号通知
    signal.Notify(r.interrupt, os.Interrupt)

    go func() {
        r.complete <- r.run()
    }()

    select {
    case err := <-r.complete:
        return err
    case <-r.timeout:
        return ErrTimeout
    }
}

func (r *JobRunner) run() error {
    for id, task := range r.tasks {
        if r.gotInterrupt() {
            return ErrInterrupt
        }
        task(id)
    }
    return nil
}

func (r *JobRunner) gotInterrupt() bool {
    select {
    case <-r.interrupt:
        signal.Stop(r.interrupt)
        return true
    default:
        return false
    }
}

package main

import (
    "fmt"
    "log"
    "os"
    "test/runner"
    "time"
)

const timeout = 3 * time.Second

func main()  {
    fmt.Println("开始运行...")

    // 初始化作业运行器
    r := runner.New(timeout)

    // 调度三个后台处理任务
    r.Add(createTask(), createTask(), createTask())

    // 启动作业运行器
    if err := r.Start(); err != nil {
        switch err {
        case runner.ErrTimeout:
            log.Println("作业程序因运行超时而终止")
            os.Exit(1)
        case runner.ErrInterrupt:
            log.Println("作业程序因系统发生中断事件而终止")
            os.Exit(2)
        }
    }
}

// 编写一个模拟后台处理任务
func createTask() func(int) {
    return func(id int) {
        log.Printf("Processor - Task #%d.", id)
        time.Sleep(time.Duration(id) * time.Second)
    }
}

--go (项目根目录 ~/Development/go)
  |--src
      |--test
          |--runner
              |--job.go
          |--runner.go
          |--go.mod