golang学习笔记之手写一个执行器

之前介绍过一个多协程的Parallelize，允许多个协程并发执行任务的函数，今天手写一个控制能力更强的的ruuner，初步实现单个协程处理，后续将继续改进为多协程处理：@H_197_2@

@H_674_4@package runner

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

@H_674_4@type Runner @H_674_4@struct {
     // interrupt chAnnel reports a signal from the
     // operating system.
     interrupt @H_674_4@chan os.Signal

     // complete chAnnel reports that processing is done.
     complete @H_674_4@chan error

     // timeout reports that time has run out.
     timeout <-@H_674_4@chan time.Time

     // tasks holds a set of functions that are executed
     // synchronously in index order.
     tasks []@H_674_4@func(int)
}

// ErrTimeout is returned when a value is received on the timeout.
@H_674_4@var ErrTimeout = errors.New("received timeout")

// ErrInterrupt is returned when an event from the OS is received.
@H_674_4@var ErrInterrupt = errors.New("received interrupt")

// New returns a new ready-to-use Runner.
@H_674_4@func New(d time.Duration) *Runner {
    @H_674_4@return &Runner{
        interrupt: @H_561_60@make(@H_674_4@chan os.Signal, 1),complete:  @H_561_60@make(@H_674_4@chan error),timeout:   time.After(d),}
}


// Add attaches tasks to the Runner. A task is a function that
// takes an int id.
@H_674_4@func (r *Runner) Add(tasks ...@H_674_4@func(int)) {
         r.tasks = append(r.tasks,tasks...)
}

// Start runs all tasks and monitors chAnnel events.
@H_674_4@func (r *Runner) Start() error {
    // We want to receive all interrupt based signals.
    // Run the different tasks on a different goroutIne.
    @H_674_4@go @H_674_4@func() {
        r.complete <- r.run()
    }()
    @H_674_4@SELEct {
    // Signaled when processing is done.
    @H_674_4@case err := <-r.complete:
        @H_674_4@return err
    // Signaled when we run out of time.
    @H_674_4@case <-r.timeout:
        @H_674_4@return ErrTimeout
       }
    }

// run executes each registered task.
@H_674_4@func (r *Runner) run() error {
    @H_674_4@for id,task := @H_674_4@range r.tasks {
             // check for an interrupt signal from the Os.
             @H_674_4@if r.goTinterrupt() {
             @H_674_4@return ErrInterrupt
            }
             // Execute the registered task.
        task(id)
        }

         @H_674_4@return nil
    }

// goTinterrupt verifies if thE interrupt signal has been issued.
@H_674_4@func (r *Runner) goTinterrupt() bool {

    @H_674_4@SELEct {
    // Signaled when an interrupt event is sent.
    @H_674_4@case <-r.interrupt:
    // Stop receiving any further signals.
        signal.Stop(r.interrupt)
        @H_674_4@return true
    // ConTinue running as normal.
    @H_674_4@default:
        @H_674_4@return false
    }
}

注解写的很详细，我简单介绍一下；Runner是一单协程的运行器，里面几个属性interrupt获取os的信号量，complete返回执行结果，timeout设置超时时间，如果超时结束运行，tasks是报错任务的。添加任务通过Add方法：将方法加入到切片中，Start方法启动任务，这里只启动一个协程，后期改进，run方式是具体执行，执行task函数goTinterrupt获取os的消息。
怎样使用呢？看下面：@H_197_2@

@H_674_4@const timeout  = 2*time.Second
func main() {
    r := runner.@H_674_4@New(timeout)
    r.Add(crateTask(),crateTask(),crateTask())
    @H_674_4@if err := r.Start();  err != nil {
        switch err {
        @H_674_4@case runner.ErrTimeout:
            fmt.Println("timeout error")
            os.@H_674_4@Exit(1)
        @H_674_4@case runner.ErrInterrupt:
            fmt.Println("interrupt error")
            os.@H_674_4@Exit(2)
        }

    }
    log.Println("end !!!!")


}

func crateTask()func(int)  {
    return func(id int) {
        fmt.Println("exec id :",id)
        time.Sleep(time.Duration(id)*time.Second)
    }
}

通过改变timeout时间可以准确的控制任务执行时间，上面2秒的例子，保证每个任务都能运行，执行结果如下： exec id : 0 exec id : 1 exec id : 2 timeout error exit status 1 如果改成5s当然能保证执行完成任务： exec id : 0 exec id : 1 exec id : 2 2017/04/11 09:59:12 end !!!!@H_197_2@ @H_301_214@