Go语言之并发示例(Runner)

简介: 这篇通过一个例子,演示使用通道来监控程序的执行时间,生命周期,甚至终止程序等。我们这个程序叫runner,我们可以称之为执行者,它可以在后台执行任何任务,而且我们还可以控制这个执行者,比如强制终止它等。

这篇通过一个例子,演示使用通道来监控程序的执行时间,生命周期,甚至终止程序等。我们这个程序叫runner,我们可以称之为执行者,它可以在后台执行任何任务,而且我们还可以控制这个执行者,比如强制终止它等。


现在开始吧,运用我们前面十几篇连载的知识,来构建我们的Runner,使用一个结构体类型就可以。


//一个执行者,可以执行任何任务,但是这些任务是限制完成的,//该执行者可以通过发送终止信号终止它
type Runner struct {
    tasks []func(int) //要执行的任务
    complete chan error //用于通知任务全部完成
    timeout <-chan time.Time //这些任务在多久内完成
    interrupt chan os.Signal //可以控制强制终止的信号
}


示例中,我们定义了一个结构体类型Runner,这个Runner包含了要执行哪些任务tasks,然后使用complete通知任务是否全部完成,不过这个执行者是有时间限制的,这就是timeout,如果在限定的时间内没有完成,就会接收到超时的通知,如果完成了就会接收到完成的通知。注意这里的timeout是单向通道,只能接收。


complete定义为error类型的通道,是为了当执行任务出现问题时返回错误的原因,如果没有出现错误,返回的是nil


此外,我们还定义了一个中断的信号,让我们可以随时的终止执行者。


有了结构体,我们接着再定义一个工厂函数New,用于返回我们需要的Runner


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


这个New函数非常简洁,可以帮我们很快的初始化一个Runnner,它只有一个参数,用来设置这个执行者的超时时间。这个超时时间被我们传递给了time.After函数,这个函数可以在tm时间后,会同伙一个time.Time类型的只能接收的单向通道,来告诉我们已经到时间了。


complete是一个无缓冲通道,也就是同步通道,因为我们要使用它来控制我们整个程序是否终止,所以它必须是同步通道,要让main goroutine等待,一致要任务完成或者被强制终止。


interrupt是一个有缓冲的通道,这样做是因为,我们可以至少接收到一个操作系统的中断信息,这样Go runtime在发送这个信号的时候不会被阻塞,如果是无缓冲的通道就会阻塞了。


系统信号是什么意思呢,比如我们在程序执行的时候按下Ctrl + C,这就是一个中断的信号,告诉程序可以强制终止了。


我们这里初始化了结构体的三个字段,而执行的任务tasks没有初始化,默认就是零nil,因为它是一个切片。但是我们的执行者Runner不能没有任务啊,既然初始化Runner的时候没有,那我们就定义一个方法,通过方法给执行者添加需要执行的任务。


//将需要执行的任务,添加到Runner里
func (r *Runner) Add(tasks ...func(int)){
    r.tasks = append(r.tasks,tasks...)
}


这个没有太多可以说明的,r.tasks就是一个切片,来存储需要执行的任务。通过内置的append函数就可以追加任务了。这里使用了可变参数,可以灵活的添加一个,甚至同时多个任务,比较方便。


到了这里我们需要的执行者Runner,如何添加任务,如何获取一个执行者,都有了,下面就开始执行者如何运行任务?如何在运行的时候强制中断任务?在这些处理之前,我们先来定义两个我们的两个错误变量,以便在接下来的代码实例中使用。


var ErrTimeOut = errors.New("执行者执行超时")
var ErrInterrupt = errors.New("执行者被中断")


两种错误类型,一个表示因为超时错误,一个表示因为被中断错误。下面我们就看看如何执行一个个任务。


//执行任务,执行的过程中接收到中断信号时,返回中断错误//如果任务全部执行完,还没有接收到中断信号,则返回nil
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
    }
}


新增的run方法也很简单,会使用for循环,不停的运行任务,在运行的每个任务之前,都会检测是否收到了中断信号,如果没有收到,则继续执行,一直到执行完毕,返回nil;如果收到了中断信号,则直接返回中断错误类型,任务执行终止。


这里注意isInterrupt函数,它在实现的时候,使用了基于select的多路复用,selectswitch很像,只不过它的每个case都是一个通信操作。那么到底选择哪个case块执行呢?原则就是哪个case的通信操作可以执行就执行哪个,如果同时有多个可以执行的case,那么就随机选择一个执行。


针对我们方法中,如果r.interrupt中接受不到值,就会执行default语句块,返回false,一旦r.interrupt中可以接收值,就会通知Go Runtime停止接收中断信号,然后返回true


这里如果没有default的话,select是会阻塞的,直到r.interrupt可以接收值为止,因为我们例子中的逻辑要求不能阻塞,所以我们使用了default


好了,基础工作都做好了,现在开始执行我们所有的任务,并且时刻监视着任务的完成,执行事件的超时。


//开始执行所有任务,并且监视通道事件
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
    }
}


signal.Notify(r.interrupt, os.Interrupt),这个是表示,如果有系统中断的信号,发给r.interrupt即可。


任务的执行,这里开启了一个groutine,然后调用run方法,结果发送给通道r.complete。最后就是使用一个select多路复用,哪个通道可以操作,就返回哪个。


到了这时候,只有两种情况了,要么任务完成;要么到时间了,任务执行超时。从我们前面的代码看,任务完成又分两种情况,一种是没有执行完,但是收到了中断信号,中断了,这时返回中断错误;一种是顺利执行完成,这时返回nil


现在把这些代码汇总一下,容易统一理解一下,所有代码如下:


package commonimport (
    "errors"
    "os"
    "os/signal"
    "time")
var ErrTimeOut = errors.New("执行者执行超时")
var ErrInterrupt = errors.New("执行者被中断")

//一个执行者,可以执行任何任务,但是这些任务是限制完成的,//该执行者可以通过发送终止信号终止它
type Runner struct {
    tasks     []func(int)      //要执行的任务
    complete  chan error       //用于通知任务全部完成
    timeout   <-chan time.Time //这些任务在多久内完成
    interrupt chan os.Signal   //可以控制强制终止的信号
}
func New(tm time.Duration) *Runner {
    return &Runner{
        complete:  make(chan error),
        timeout:   time.After(tm),
        interrupt: make(chan os.Signal, 1),
    }
}
//将需要执行的任务,添加到Runner里
func (r *Runner) Add(tasks ...func(int)) {
    r.tasks = append(r.tasks, tasks...)
}
//执行任务,执行的过程中接收到中断信号时,返回中断错误//如果任务全部执行完,还没有接收到中断信号,则返回nil
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 (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
    }
}


这个common包里的Runner我们已经开发完了,现在我们写个例子试试它。


package main
import (
    "flysnow.org/hello/common"
    "log"
    "time"
    "os")
func main() {
    log.Println("...开始执行任务...")

    timeout := 3 * time.Second
    r := common.New(timeout)

    r.Add(createTask(), createTask(), createTask())
    if err:=r.Start();err!=nil{
            switch err { 
            case common.ErrTimeOut:
                    log.Println(err)
                    os.Exit(1)        
            case common.ErrInterrupt:
                   log.Println(err)
                   os.Exit(2)
        }
    }
    log.Println("...任务执行结束...")
}
func createTask() func(int) {
    return func(id int) {
        log.Printf("正在执行任务%d", id)
        time.Sleep(time.Duration(id)* time.Second)
    }
}


例子非常简单,定义任务超时时间为 3 秒,添加 3 个生成的任务,每个任务都是打印一个正在执行哪个任务,然后休眠一段时间。


调用r.Start()开始执行任务,如果一切都正常的话,返回nil,然后打印出...任务执行结束...,不过我们例子中,因为超时时间和任务的设定,结果是执行超时的。


2017/04/15 22:17:55 ...开始执行任务...
2017/04/15 22:17:55 正在执行任务0
2017/04/15 22:17:55 正在执行任务1
2017/04/15 22:17:56 正在执行任务2
2017/04/15 22:17:58 执行者执行超时


如果我们把超时时间改为 4 秒或者更多,就会打印...任务执行结束...。这里我们还可以测试另外一种系统中断情况,在终端里运行程序后,快速不停地按Ctrl + C,就可以看到执行者被中断的打印输出信息了。


到这里,这篇文章已经要收尾了,这个例子中,我们演示使用通道通信、同步等待,监控程序等。


此外这个执行者也是一个很不错的模式,比如我们写好之后,交给定时任务去执行即可,比如cron,这个模式我们还可以扩展,更高效率的并发,更多灵活的控制程序的生命周期,更高效的监控等,这个大家自己可以试试,基于自己的需求修改就可以了。


目录
打赏
0
0
0
0
9
分享
相关文章
Go语言中的并发编程:深入理解与实践###
探索Go语言在并发编程中的独特优势,揭秘其高效实现的底层机制。本文通过实例和分析,引导读者从基础到进阶,掌握Goroutines、Channels等核心概念,提升并发处理能力。 ###
Go语言中的并发编程:掌握goroutines和channels####
本文深入探讨了Go语言中并发编程的核心概念——goroutine和channel。不同于传统的线程模型,Go通过轻量级的goroutine和通信机制channel,实现了高效的并发处理。我们将从基础概念开始,逐步深入到实际应用案例,揭示如何在Go语言中优雅地实现并发控制和数据同步。 ####
Go语言中的并发编程与通道(Channel)的深度探索
本文旨在深入探讨Go语言中并发编程的核心概念和实践,特别是通道(Channel)的使用。通过分析Goroutines和Channels的基本工作原理,我们将了解如何在Go语言中高效地实现并行任务处理。本文不仅介绍了基础语法和用法,还深入讨论了高级特性如缓冲通道、选择性接收以及超时控制等,旨在为读者提供一个全面的并发编程视角。
100 50
Go语言中的并发编程:深入理解与实践####
本文旨在为读者提供一个关于Go语言并发编程的全面指南。我们将从并发的基本概念讲起,逐步深入到Go语言特有的goroutine和channel机制,探讨它们如何简化多线程编程的复杂性。通过实例演示和代码分析,本文将揭示Go语言在处理并发任务时的优势,以及如何在实际项目中高效利用这些特性来提升性能和响应速度。无论你是Go语言的初学者还是有一定经验的开发者,本文都将为你提供有价值的见解和实用的技巧。 ####
Go语言的并发编程:深入理解与实践####
本文旨在探讨Go语言在并发编程方面的独特优势及其实现机制,通过实例解析关键概念如goroutine和channel,帮助开发者更高效地利用Go进行高性能软件开发。不同于传统的摘要概述,本文将以一个简短的故事开头,引出并发编程的重要性,随后详细阐述Go语言如何简化复杂并发任务的处理,最后通过实际案例展示其强大功能。 --- ###
Go语言中的并发编程:深入理解goroutines和channels####
本文旨在探讨Go语言中并发编程的核心概念——goroutines和channels。通过分析它们的工作原理、使用场景以及最佳实践,帮助开发者更好地理解和运用这两种强大的工具来构建高效、可扩展的应用程序。文章还将涵盖一些常见的陷阱和解决方案,以确保在实际应用中能够避免潜在的问题。 ####
Go语言中的并发编程:掌握goroutine和channel的艺术####
本文深入探讨了Go语言在并发编程领域的核心概念——goroutine与channel。不同于传统的单线程执行模式,Go通过轻量级的goroutine实现了高效的并发处理,而channel作为goroutines之间通信的桥梁,确保了数据传递的安全性与高效性。文章首先简述了goroutine的基本特性及其创建方法,随后详细解析了channel的类型、操作以及它们如何协同工作以构建健壮的并发应用。此外,还介绍了select语句在多路复用中的应用,以及如何利用WaitGroup等待一组goroutine完成。最后,通过一个实际案例展示了如何在Go中设计并实现一个简单的并发程序,旨在帮助读者理解并掌
Go语言中的并发编程:掌握goroutine与通道的艺术####
本文深入探讨了Go语言中的核心特性——并发编程,通过实例解析goroutine和通道的高效使用技巧,旨在帮助开发者提升多线程程序的性能与可靠性。 ####
Go语言中的并发编程:掌握goroutines和channels####
本文深入探讨了Go语言中并发编程的核心概念,重点介绍了goroutines和channels的工作原理及其在实际开发中的应用。文章通过实例演示如何有效地利用这些工具来编写高效、可维护的并发程序,旨在帮助读者理解并掌握Go语言在处理并发任务时的强大能力。 ####
|
2月前
|
Go 语言以其高效的并发编程能力著称,主要依赖于 goroutines 和 channels 两大核心机制
Go 语言以其高效的并发编程能力著称,主要依赖于 goroutines 和 channels 两大核心机制。本文介绍了这两者的概念、用法及如何结合使用,实现任务的高效并发执行与数据的安全传递,强调了并发编程中的注意事项,旨在帮助开发者更好地掌握 Go 语言的并发编程技巧。
40 2