控制程序的生命周期

这是一篇使用并发和通道来实现控制程序生命周期的并发模式示例，该示例演示了如何控制程序在规定时间段内的执行，并可以手动中断来终止程序的运行。

功能

展示如何通过通道来监视程序的执行时间，如果程序执行时间过长，也可以终止程序

使用场景

当需要调度后台处理任务的时候，这种模式会很有用。该程序可能会作为 cron 作业执行，或者在基于定时任务的云环境 (如 iron.io) 里执行。

实现思路

1. 创建一个执行者 runner, 给 runner 设置一个超时时间 timeout 和任务切片 tasks，然后遍历执行 tasks 所有任务。

2. 设置一个存储中断信号的字段 interrupt，通过 interrupt 来判断是否已经中断程序。

3. 设置一个记录每个任务执行错误结果的字段 complete，监听 complete, 判断是那种错误类型，然后做相应的处理。

4. 执行所有任务，并监听不同错误码，执行不同的业务逻辑。

实现详情

首先我们需要声明一个 runner 结构

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

runner 中包括任务切片 tasks, tasks 是一个存储 func(int) 类型的切片，后面会遍历 tasks 来进行处理任务。

timeout 字段是一个存放超时时间的只读的通道，通过该字段来判断任务执行是否超时。

interrupt 字段是存放 os.Signal 类型的通道，接收到来自终端的中断信号会存放在该字段中。

complete 字段是存放任务执行的错误结果，如果没有错误则是 nil。

有了 runner 执行者这个结构后，我们可以声明一个 New 工厂函数来创建 runner 类型的对象，并初始化需要的字段。

// 工厂函数创建 runner
func New(timeout time.Duration) *runner {
  return &runner{
    timeout: time.After(timeout),
    interrupt: make(chan os.Signal),
    complete: make(chan error),
  }
}

创建 runner 的时候，我们需要传入一个 time.Duration 类型的参数，然后内部调用 time.After() 这个函数来返回一个time.Time 类型的只读通道。interrupt 和 complete 字段正常初始化即可。tasks 默认是空切片（表示还没有任何任务）。

有了一个 runner 执行者对象后，在执行任务之前我们需要给 runner 的增加任务，那我们需要写一个给 runner 增加任务的方法。

// 增加任务
// 可变参数 ...func(int) 表示参数可以是多个参数
func (r *runner) Add(tasks ...func(int)) {
  // 使用 tasks... 解构 tasks
  r.tasks = append(r.tasks, tasks...)
}

增加任务方法 Add 接收一个参数类型为 func(int) 的可变参数，可变参数意味着参数的数量是可变的，可以是单个，也可以是多个。

当调用 Add 方法后，就会把传入的参数赋值给 runner 类型对象的 tasks 字段，此时 runner 类型对象就有任务了。

因为任务执行过程会有多个错误值，比如超时错误和中断错误，所以我们先定义两个错误变量，以备后面使用。

// 错误类型
var (
  ErrTimeout = errors.New("超时错误")
  ErrInterrupt = errors.New("程序中断错误")
)

接下来就是任务的执行。

// 执行任务
func (r *runner) Run() error {
  for id, task := range r.tasks {
    // 判断是否已经中断程序
    if r.isInterrupt() {
      return ErrInterrupt
    }
    task(id)
  }
  return nil
}

执行任务就是遍历runner 对象中的 tasks 的所有任务，然后执行每一个任务即可，但是在执行任务之前，需要判断是否已经中断了程序。如果已经中断了程序，则直接返回中断错误 ErrInterrupt。

以下是判断程序中断的方法

// 判断是否中断程序
func (r *runner) isInterrupt() bool {
  select {
  case <- r.interrupt:
    signal.Stop(r.interrupt)
    return true
  default:
    return false
  }
}

该方法是 runner 类型的一个方法，方法内使用了 select 多路复用来进行监听

interrupt 通道是否有中断信号，如果监听到有中断信号，则任务是用户中断了程序，此时会调用 signal.Stop() 方法 中断程序，然后返回 true ，表示程序已经被中断。

接下来，我们的实现一个方法来整合整个任务执行的流程，包括任务的执行，中断和超时的监听。

// 执行所有任务，并监听通道事件
func (r *runner) 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
  }
}

Start 方法中，会接收所有的中断信号，放在 interrupt 通道中，然后调用 Run 方法来执行所有任务，并把执行的结果存放到 complete 通道中，最后通过 select 多路复用方式监听 complete 通道和 timeout 通道中的消息，一旦有错误就返回错误码。

执行者调用 Start 方法后拿到错误码，执行自己的业务逻辑，如果没有错误码返回则表示所有任务在规定的超时时间内成功执行了所有任务。

目前所有任务的执行，错误码监听等工作已经全部完成。

接下来我们创建一个 runner 对象来验证一下程序。

func main() {
  timeout := 2 * time.Second
  runner := New(timeout)
  runner.Add(CreateTask(), CreateTask(), CreateTask())
​
  if err := runner.Start(); err != nil {
    switch err {
    case ErrInterrupt:
      fmt.Println(ErrInterrupt)
    case ErrTimeout:
      fmt.Println(ErrTimeout)
    }
  }
  fmt.Println("程序结束")
}

因为 Add 方法参数要求是一个传入 int 类型的函数，所以为了方便创建任务，我们声明一个使用了闭包的 CreateTask 函数来返回任务函数。

// 创建任务
func CreateTask() func(int)  {
  return func(id int) {
    fmt.Println("正在执行 Task ", id)
    // 模拟任务执行
    time.Sleep(time.Duration(id) * time.Second)
  }
}

到目前为止所有的代码实现已经全部编写完成

以下是完整的示例代码

package main
​
import (
  "errors"
  "fmt"
  "os"
  "os/signal"
  "time"
)
​
type runner struct {
  tasks []func(int)
  timeout <-chan time.Time
  interrupt chan os.Signal
  complete chan error
}
​
// 工厂函数创建 runner
func New(timeout time.Duration) *runner
 {
  return &runner{
    timeout: time.After(timeout),
    interrupt: make(chan os.Signal),
    complete: make(chan error),
  }
}
​
// 错误类型
var (
  ErrTimeout = errors.New("超时错误")
  ErrInterrupt = errors.New("程序中断错误")
)
​
// 执行任务
func (r *runner) Run() error {
  for id, task := range r.tasks {
    // 判断是否已经中断程序
    if r.isInterrupt() {
      return ErrInterrupt
    }
    task(id)
  }
  return nil
}
​
// 判断是否中断程序
func (r *runner) isInterrupt() bool {
  select {
  case <- r.interrupt:
    signal.Stop(r.interrupt)
    return true
  default:
    return false
  }
}
​
// 增加任务
// 可变参数 ...func(int) 表示参数可以是多个参数
func (r *runner) Add(tasks ...func(int)) {
  // 使用 tasks... 解构 tasks
  r.tasks = append(r.tasks, tasks...)
}
​
// 执行所有任务，并监听通道事件
func (r *runner) 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 CreateTask() func(int)  {
  return func(id int) {
    fmt.Println("正在执行 Task ", id)
    time.Sleep(time.Duration(id) * time.Second)
  }
}
​
​
func main() {
  timeout := 2 * time.Second
  runner := New(timeout)
  runner.Add(CreateTask(), CreateTask(), CreateTask())
​
  if err := runner.Start(); err != nil {
    switch err {
    case ErrInterrupt:
      fmt.Println(ErrInterrupt)
    case ErrTimeout:
      fmt.Println(ErrTimeout)
    }
  }
  fmt.Println("程序结束")
}

一起精进Go技术， 关注公众号：陆贵成