go并发之路(五)——runner

发表于

本文同时参与「掘力星计划」，赢取创作大礼包，挑战创作激励金。

runner总体设计

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
go复制代码// Package runner 管理进程的运行和生命周期。
package runner

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

//Runner 在给定的超时时间内运行一组任务，并且可以在操作系统中断时关闭。
type Runner struct {
   // interrupt通道会传送操作系统发送来的信号 
   interrupt chan os.Signal

   // complete 通道会发来协程完成的信号
   complete chan error

   // timeout 会发来超时信号
   timeout <-chan time.Time

   // 任务包含一组按索引顺序同步执行的函数。
   tasks []func(int)
}

// 当在超时通道上接收到一个值时，返回 ErrTimeout。
var ErrTimeout = errors.New("received timeout")

// 当接收到来自操作系统的事件时，会返回 ErrInterrupt。
var ErrInterrupt = errors.New("received interrupt")

// New 返回一个新的准备被使用的Runner。
func New(d time.Duration) *Runner {
   return &Runner{
      interrupt: make(chan os.Signal, 1),
      complete:  make(chan error),
      timeout:   time.After(d),
   }
}

从总体设计上，我们可以看出runner具备以下功能：

  • 程序如果在分配的超时时间内完成工作，可以正常终止;
  • 程序没有及时完成工作，就会选择“自杀”;
  • 程序如果接收到操作系统发送的中断事件，程序立刻试图清理状态并停止工作。

runner细节设计

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
go复制代码// 添加任务到 Runner。 任务是一个接受 int整数型变量的函数。
func (r *Runner) Add(tasks ...func(int)) {
   r.tasks = append(r.tasks, tasks...)
}

// Start函数运行所有任务并监听通道事件。
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
   }
}

// run函数执行每个注册的任务。
func (r *Runner) run() error {
   for id, task := range r.tasks {
      // 检查来自操作系统的中断信号。
      if r.gotInterrupt() {
         return ErrInterrupt
      }

      // 执行注册的任务。
      task(id)
   }

   return nil
}

// gotInterrupt函数验证是否已发出中断信号。
func (r *Runner) gotInterrupt() bool {
   select {
   // 中断事件被发送时发出信号。
   case <-r.interrupt:
      // 停止接收任何进一步的信号。
      signal.Stop(r.interrupt)
      return true

   // 像往常一样正常运行
   default:
      return false
   }
}

以上代码分别是任务的添加，以及任务开始，执行，中断步骤，共同构成了一个完整的生命周期。

举个例子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
go复制代码package main
import (
   "log"
   "os"
   "time"

   "github.com/goinaction/code/chapter7/patterns/runner"
)

// timeout 规定了必须在多少秒内处理完成
const timeout = 3 * time.Second

// main 是程序的入口
func main() {
   log.Println("Starting work.")

   // 为本次执行分配超时时间
   r := runner.New(timeout)

   // 加入要执行的任务
   r.Add(createTask(), createTask(), createTask(), createTask(), createTask(),createTask())

   // 执行任务并处理结果
    if err := r.Start(); err != nil {
       log.Println(err.Error())
    }
   log.Println("Process ended.")
}

// createTask 返回一个根据 id
// 休眠指定秒数的示例任务
// 用来模拟正在进行工作
func createTask() func(int) {
   return func(id int) {
      log.Printf("Processor - Task #%d.", id)
      time.Sleep(time.Duration(id) * time.Second)
   }
}

模拟代码简单的设置了任务并展示了任务执行成功以及超时会出现的现象

1
2
3
4
5
6
yaml复制代码2021/10/18 01:09:39 Starting work.
2021/10/18 01:09:39 Processor - Task #0.
2021/10/18 01:09:39 Processor - Task #1.
2021/10/18 01:09:40 Processor - Task #2.
2021/10/18 01:09:42 received timeout
2021/10/18 01:09:42 Process ended.

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%