更便捷的goroutine控制利器- Context

简介: 在本文中,我首先会介绍context是什么,它有什么作用,以及如何使用,其中还会参杂一点个人的理解,以及部分源码的了解。What are you waiting for?

Context:


来自官方文档


Context包定义了上下文类型,该类型在API边界之间以及进程之间传递截止日期,取消信号和其他请求范围的值


对服务器的传入请求应创建一个Context,而对服务器的传出调用应接受一个Context。


它们之间的函数调用链必须传播Context,可以选择将其替换为使用WithCancel,WithDeadline,WithTimeout或WithValue创建的派生Context。取消上下文后,从该上下文派生的所有上下文也会被取消。


WithCancel,WithDeadline和WithTimeout函数采用Context(父级)并返回派生的Context(子级)和CancelFunc。调用CancelFunc会取消该子代及其子代,删除父代对该子代的引用,并停止所有关联的计时器。未能调用CancelFunc会使子代及其子代泄漏,直到父代被取消或计时器触发。审核工具检查所有控制流路径上是否都使用了CancelFuncs。


使用上下文的程序应遵循以下规则,以使各个包之间的接口保持一致,并使静态分析工具可以检查上下文传播:


不要将上下文存储在结构类型中;而是将上下文明确传递给需要它的每个函数。Context应该是第一个参数,通常命名为ctx:


func DoSomething(ctx context.Context, arg Arg) error {
//  ... use ctx ...
}


即使函数允许,也不要传递nil Context。如果不确定使用哪个上下文,请传递context.TODO


仅将上下文值用于传递过程和API的请求范围数据,而不用于将可选参数传递给函数。

可以将相同的上下文传递给在不同goroutine中运行的函数。上下文可以安全地被多个goroutine同时使用


巴拉巴拉,说了一大堆,反正我一句没懂,当然我知道context是干嘛的,(尬~,不小心暴露了,学渣的本质),说说我的理解以及使用建议


  • 对服务器的传入请求应创建一个Context,而对服务器的传出响应也应接受一个Context。


  • 函数调用链必须传播Context,也可以选择将其替换为使用WithCancel,WithDeadline,WithTimeout或WithValue创建的派生Context(也就是子类context)。取消上下文后,从该上下文派生的所有上下文也会被取消


  • Context 不要放在结构体中,要以参数的方式传递。


  • Context 作为函数的参数时,要放在第一位,也就是第一个参数。


  • 要使用 context.Background 函数生成根节点的 Context,也就是最顶层的 Context。


  • Context 传值要传递必须的值,而且要尽可能地少,不要什么都传。


  • Context 多协程安全,可以在多个协程中放心使用。


go Context定义


Context 是Go 1.7 标准库引入 的标准库,中文译作“上下文”,准确说它是 goroutine 的上下文,包含 goroutine 的运行状态、环境、现场等信息。


使用context,我们可以轻松优雅的做到取消goroutine超时时间,运行截止时间k-v存储等。它是并发安全的


随着 context 包的引入,标准库中很多接口因此加上了 context 参数,例如 database/sql 包。context 几乎成为了并发控制和超时控制的标准做法。


context.Context 类型的值可以协调多个 groutine 中的代码执行“取消”操作,并且可以存储键值对。最重要的是它是并发安全的。


与它协作的 API 都可以由外部控制执行“取消”操作,例如:取消一个 HTTP 请求的执行。


止于这些么?当然 不止,还有更多的骚操作,接下来让我们一起拿下它吧。


引入


为什么需要使用context,理由一


一个协程启动后,大部分情况需要等待里面的代码执行完毕,然后协程会自行退出。

但需要让协程提前退出怎么办呢?


下面我们以一个小的示例,来逐渐了解context的妙用之一吧


package main
import (
    "fmt"
    "sync"
    "time"
)
func monitor(name string) {
    //开启for select循环,j进行后台监控
    for {
        select {
        default:
            fmt.Printf("Time: %v 监控者:%s, 正在监控...\n", time.Now().Unix(), name)
        }
        // sleep 1 second
        time.Sleep(time.Second * 5)
        fmt.Printf("%s 监控完成,一切正常,请指示 over...\n", name)
    }
}
func main() {
    var wg sync.WaitGroup
    // Define waiting group
    wg.Add(1)
    go func() {
        // Execution complete
        defer wg.Done()
        monitor("天眼")
    }()
    //Exit after waiting
    wg.Wait()
}


我们在这里实现了一个基本的groutine执行的case


我们定义了等待组wait group,防止协程提前退出。关于wait group可参考上一篇文章,golang并发控制的心应手。


他会周期性的运行,不断打印监控信息,例如


640 (3).jpg


那么我们完成上述的那个需求提前退出,那么该怎么办呢?其中一个方法就是定义一个全局的sign,其他地方可以通过修改这个sign发出停止监控的指令。然后在协程中先检查这个变量,如果发现被通知关闭就停止监控,退出当前协程。从而实现可控制提前退出。示例代码如下


package main
import (
    "fmt"
    "sync"
    "time"
)
func monitor1(signCh chan bool, MonitoringPeriod time.Duration, name string, ) {
    //开启for select循环,一直后台监控
    for {
        select {
        case <-signCh:
            fmt.Println(name, "停止指令已收到,停止...")
            return
        default:
            fmt.Printf("Time: %v [监控者]:%s, 正在监控...\n", time.Now().Unix(), name)
        }
        time.Sleep(MonitoringPeriod * time.Second)
    }
}
func main() {
    var wg sync.WaitGroup
    signCh := make(chan bool) //sign 用来停止监控
    const MonitoringTime, MonitoringPeriod = 20, 2
    wg.Add(1)
    go func() {
        defer wg.Done()
        monitor1(signCh, MonitoringPeriod, "天眼")
    }()
    time.Sleep(MonitoringTime * time.Second) //实施监控时间
    signCh <- true                           //发停止指令
    wg.Wait()
}


这样我们就实现了,可控制话的groutine退出,但如果在新增几个定期的任务功能,那该如何是好?


管他的,我们先把这个弄懂了先。老夫先干为敬。首先我们先看程序运行图,如下


640 (4).jpg


这个示例是使用 select+channel 的方式改造,实现了通过 channel 发送指令让监控狗停止,进而达到协程退出的目的。


首先我们定义了sync.WaitGroup,防止gorontine提前退出。signCh,他是一个bool值类型channel,用于发送sign后续的退出。


MonitoringTime,MonitoringPeriod,监控时间与监控周期。second。


然后创建goroutine执行select+channel。


Go Context 初试体验


为 函数增加 signCh 参数,用于接收停止指令;

在 main 函数中,声明用于停止的 signCh,传递给 monitor1 函数,然后通过 signCh<-true 发送停止指令让协程退出。


通过 select+channel 让协程退出的方式比较优雅,以下几个问题也随之凸显

但如果我们希望做到同时取消很多个协程呢?


如果是定时取消协程又该怎么办?


这时候 select+channel 的局限性就凸现出来了,即使定义了多个 channel 解决问题,当然这个方式是可行的,但代码逻辑也会非常复杂、难以维护。


要解决这种复杂的协程问题,必须有一种可以跟踪协程的方案,只有跟踪到每个协程,才能更好地控制它们,这种方案就是 Go 语言标准库为我们提供的 Context,接下来我们体验一下它的强大之处吧。


package main
import (
    "context"
    "fmt"
    "sync"
    "time"
)
func main() {
    var wg sync.WaitGroup
    const MonitoringTime, MonitoringPeriod = 20, 2
    wg.Add(1)
    // 定义一个等待的 `context`
    ctx, stop := context.WithCancel(context.Background())
    go func() {
        defer wg.Done()
        monitor2(ctx, MonitoringPeriod, "天眼")
    }()
    time.Sleep(MonitoringTime * time.Second) //先监控5秒
    stop()                                   //发停止指令
    wg.Wait()
}
func monitor2(ctx context.Context, MonitoringPeriod time.Duration, name string, ) {
    //开启for select循环,一直后台监控
    for {
        select {
        case <-ctx.Done():
            fmt.Println(name, "停止指令已收到,停止...")
            return
        default:
            fmt.Printf("Time: %v [监控者]:%s, 正在监控...\n", time.Now().Unix(), name)
        }
        time.Sleep(MonitoringPeriod * time.Second)
    }
}


是不是很优雅呢?确实如此,那么为什么也可以达到上面使用channel,的效果呢。那么我们去看一下它的具体实现部分呢,


`WithCancel`


以下是WithCancel:具体实现部分代码


WithCancel:返回具有新的“完成”通道的父级副本。当调用返回的cancel函数或关闭父上下文的Done通道时(以先发生的为准),将关闭返回的上下文的Done通道。取消此上下文将释放与其关联的资源,因此在此上下文中运行的操作完成后,代码应立即调用cancel。


func WithCancel(parent Context) (ctx Context, cancel CancelFunc) {
    if parent == nil {
        panic("cannot create context from nil parent")
    }
    c := newCancelCtx(parent)
    propagateCancel(parent, &c)
    return &c, func() { c.cancel(true, Canceled) }
}


除了WithCancel,之外还有WithDeadline,WithTimeout,WithValue,首先我们来继续看看WithDeadline具体实现,以及使用技巧吧


`WithTimeout`


func WithDeadline(parent Context, d time.Time) (Context, CancelFunc) {
    if parent == nil {
        panic("cannot create context from nil parent")
    }
    if cur, ok := parent.Deadline(); ok && cur.Before(d) {
        // The current deadline is already sooner than the new one.
        return WithCancel(parent)
    }
    c := &timerCtx{
        cancelCtx: newCancelCtx(parent),
        deadline:  d,
    }
    propagateCancel(parent, c)
    dur := time.Until(d)
    if dur <= 0 {
        c.cancel(true, DeadlineExceeded) // deadline has already passed
        return c, func() { c.cancel(false, Canceled) }
    }
    c.mu.Lock()
    defer c.mu.Unlock()
    if c.err == nil {
        c.timer = time.AfterFunc(dur, func() {
            c.cancel(true, DeadlineExceeded)
        })
    }
    return c, func() { c.cancel(true, Canceled) }
}
func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc) {
    return WithDeadline(parent, time.Now().Add(timeout))
}


取消此上下文将释放与之关联的资源,因此在此上下文中运行的操作完成后,代码应立即调用cancel:


来看一下具体如何使用吧,示例如下


package main
import (
    "context"
    "fmt"
    "time"
)
func main() {
    // 创建一个子节点的context,3秒后自动超时
    const MonitoringTime, MonitoringPeriod = 20, 2
    ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
    go func() {
        monitor4(ctx, MonitoringPeriod, "天眼")
        fmt.Println("退出时间",time.Now().Unix())
    }()
    time.Sleep(MonitoringTime * time.Second)
    cancel()
}
func monitor4(ctx context.Context, MonitoringPeriod time.Duration, name string, ) {
    //开启for select循环,一直后台监控
    for {
        select {
        case <-ctx.Done():
            fmt.Println(name, "停止指令已收到,停止...")
            return
        default:
            fmt.Printf("Time: %v [监控者]:%s, 正在监控...\n", time.Now().Unix(), name)
            time.Sleep(MonitoringPeriod * time.Second)
        }
    }
}


以上会有两种情况发生退出,


一、程序main退出,全局退出


二、我们定义的timeout退出


他们的基本性质与使用我们就简单的过了一遍,下面让我们来个小结。


WithCancel(parent Context):生成一个可取消的 Context。


WithDeadline(parent Context, d time.Time):生成一个可定时取消的 Context,参数 d 为定时取消的具体时间。


WithTimeout(parent Context, timeout time.Duration):生成一个可超时取消的 Context,参数 timeout 用于设置多久后取消


WithValue(parent Context, key, val interface{}):生成一个可携带 key-value 键值对的 Context。


是不是发现,其实也没有那么难呢?当然,它本来就很简单,接下来我们来点更刺激的,同时取消多goroutine,啥也不说了,上~


package main
import (
    "context"
    "fmt"
    "strconv"
    "sync"
    "time"
)
var wg sync.WaitGroup
func main() {
    const MonitoringTime, MonitoringPeriod = 20, 2
    wg.Add(1)
    ctx, stop := context.WithCancel(context.Background())
    for i := 0; i < 3; i++ {
        go monitor6(ctx, MonitoringPeriod, strconv.Itoa(i))
    }
    time.Sleep(MonitoringTime * time.Second) //先监控5秒
    stop()                                   //发停止指令
    wg.Wait()
}
func monitor6(ctx context.Context, MonitoringPeriod time.Duration, name string, ) {
    defer wg.Done()
    //开启for select循环,一直后台监控
    for {
        select {
        case <-ctx.Done():
            fmt.Println(name, "停止指令已收到,停止...")
            return
        default:
            fmt.Printf("Time: %v [监控者]:天眼%s, 正在监控...\n", time.Now().Unix(), name)
        }
        time.Sleep(MonitoringPeriod * time.Second)
    }
}


Time: 1612948086 [监控者]:天眼0, 正在监控…

Time: 1612948086 [监控者]:天眼1, 正在监控…

Time: 1612948086 [监控者]:天眼2, 正在监控…


… …


Time: 1612948104 [监控者]:天眼2, 正在监控…

Time: 1612948104 [监控者]:天眼0, 正在监控…

Time: 1612948104 [监控者]:天眼1, 正在监控…


2 停止指令已收到,停止…

1 停止指令已收到,停止…

0 停止指令已收到,停止…


你以为这样就完了么,这只是一个小的case,它还可以管理子节点。其管理与树形结构十分的相似。


除此之外还可以传递值,接下来让我们来看看吧


package main
import (
    "context"
    "fmt"
    "sync"
    "time"
)
func main() {
    var wg sync.WaitGroup
    ctx, stop := context.WithCancel(context.Background())
    ctxVal := context.WithValue(ctx, "user", "payne")
    wg.Add(1)
    go func() {
        defer wg.Done()
        getValue(ctxVal)
    }()
    time.Sleep(3)
    stop()
    wg.Wait()
}
func getValue(ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            fmt.Println("exit")
            return
        default:
            user := ctx.Value("user")
            fmt.Println("【获取用户】", "用户为:", user)
            time.Sleep(1 * time.Second)
        }
    }
}


输出如下


【获取用户】 用户为:payne

exit…


总结:


Context为我们主要定义四种方法

WithDeadline,WithTimeout,WithValue,WithCancel,从而达到控制goroutine的目的,但却不仅限于我们以上介绍的那样(只介绍了一层,其实可以是多层。形成多对多的关系),它更深层次的使用你可以想象成多叉树的情况。


context,这一篇就暂且完成啦,期待下一篇。并发模式


并发模式,故名思义。他与设计模式一样,即使用goroutine并发的一些总结。


我将与你探讨


  • Goroutine WorkPool:让我们随所欲的控制创建gototine的数量,且复用。


  • Pipeline 模式,他像工厂流水线一般,我们将是这将其拆分归并


  • 扇出扇入模式,在pipline的基础上对耗时较长的进行处理


  • Futures 模式,Pipeline 流水线模式中的工序是相互依赖的,但是在我们的实际需求中,也有大量的任务之间相互独立、没有依赖,所以为了提高性能,这些独立的任务就可以并发执行。
目录
相关文章
|
5天前
|
数据采集 人工智能 自然语言处理
3分钟采集134篇AI文章!深度解析如何通过云无影AgentBay实现25倍并发 + LlamaIndex智能推荐
结合阿里云无影 AgentBay 云端并发采集与 LlamaIndex 智能分析,3分钟高效抓取134篇 AI Agent 文章,实现 AI 推荐、智能问答与知识沉淀,打造从数据获取到价值提炼的完整闭环。
390 93
|
6天前
|
域名解析 人工智能
【实操攻略】手把手教学,免费领取.CN域名
即日起至2025年12月31日,购买万小智AI建站或云·企业官网,每单可免费领1个.CN域名首年!跟我了解领取攻略吧~
|
5天前
|
SQL 人工智能 自然语言处理
Geo优化SOP标准化:于磊老师的“人性化Geo”体系如何助力企业获客提效46%
随着生成式AI的普及,Geo优化(Generative Engine Optimization)已成为企业获客的新战场。然而,缺乏标准化流程(Geo优化sop)导致优化效果参差不齐。本文将深入探讨Geo专家于磊老师提出的“人性化Geo”优化体系,并展示Geo优化sop标准化如何帮助企业实现获客效率提升46%的惊人效果,为企业在AI时代构建稳定的流量护城河。
391 156
Geo优化SOP标准化:于磊老师的“人性化Geo”体系如何助力企业获客提效46%
|
5天前
|
数据采集 缓存 数据可视化
Android 无侵入式数据采集:从手动埋点到字节码插桩的演进之路
本文深入探讨Android无侵入式埋点技术,通过AOP与字节码插桩(如ASM)实现数据采集自动化,彻底解耦业务代码与埋点逻辑。涵盖页面浏览、点击事件自动追踪及注解驱动的半自动化方案,提升数据质量与研发效率,助力团队迈向高效、稳定的智能化埋点体系。(238字)
272 158
|
13天前
|
机器人 API 调度
基于 DMS Dify+Notebook+Airflow 实现 Agent 的一站式开发
本文提出“DMS Dify + Notebook + Airflow”三位一体架构,解决 Dify 在代码执行与定时调度上的局限。通过 Notebook 扩展 Python 环境,Airflow实现任务调度,构建可扩展、可运维的企业级智能 Agent 系统,提升大模型应用的工程化能力。