经营范围 网站建设网络运营课程培训班
题目
手撕 对无序的切片查询指定数 使用context进行子协程的销毁 并且进行超时处理。
全局变量定义
var (startLoc = int64(0) // --- 未处理切片数据起始位置endLoc = int64(0) // --- 切片数据右边界 避免越界offset = int64(0) // --- 根据切片和协程数量 在主线程 动态设置target = 42 // --- 设置的目标值mu sync.Mutex // --- 避免并发冲突使用的全局锁
)1.并发处理
1.1 使用atomic原子操作
使用CAS操作解决并发问题（不使用锁） 效率上和使用全局锁在 100000 上几乎没差别
// --- 使用atomic原子操作
start = atomic.LoadInt64(&startLoc)
end = start + offset
if end > endLoc {end = endLoc
}
// 应该不会出现ABA问题
if ok := atomic.CompareAndSwapInt64(&startLoc, start, end); ok == false {continue
}1.2 使用全局锁
mu.Lock()
start = startLoc
end = start + offset
startLoc = end
mu.Unlock()
if start >= endLoc {return
}
if end > endLoc {end = endLoc
}1.3主线程手动切片全部代码
package mainimport ("context""fmt""sync""sync/atomic""time"
)func find(nums []int, ctx context.Context, wg *sync.WaitGroup, target int, start, end int64) {defer wg.Done()for {select {case <-ctx.Done():// 如果接收到取消信号，退出协程returndefault:for i := start; i < end; i++ {if nums[i] == target {// 使用 atomic 以确保线程安全atomic.StoreInt32(&valid, 1)return}}return}}
}var valid int32func main() {sliceLen := int64(1000000)// 创建一个背景上下文和一个取消功能ctx := context.Background()// 假设 ddl 是一个固定的截止时间ddl := time.Now().Add(10 * time.Second) // 假设 5 秒钟后超时newCtx, cancel := context.WithDeadline(ctx, ddl)// 创建一个较大的切片 nums 并初始化nums := make([]int, sliceLen)// 初始化切片为随机数据，例如从 1 到 100，值为42的即为目标for i := 0; i < len(nums); i++ {nums[i] = i}offset := sliceLen / 10startLoc := int64(0)startTime := time.Now()// 使用 WaitGroup 来等待所有协程完成var wg sync.WaitGroup// 启动多个协程进行查找for i := 0; i < 10; i++ {wg.Add(1)go find(nums, newCtx, &wg, 42, startLoc, startLoc+offset)startLoc = startLoc + offset}// 等待结果go func() {wg.Wait()cancel() // 等待所有协程结束后，调用 cancel}()// 检查结果select {case <-newCtx.Done():if atomic.LoadInt32(&valid) == 1 {fmt.Println("Found target!")} else {fmt.Println("Timeout or not found.")}}duration := time.Since(startTime)fmt.Printf("程序运行时间: %s\n", duration)
}
1.4 采取锁处理 & 原子操作 全部代码
package mainimport ("context""fmt""sync""sync/atomic""time"
)var (startLoc = int64(0)endLoc = int64(0)offset = int64(0)target = 42mu sync.Mutex
)func find(nums []int, ctx context.Context, wg *sync.WaitGroup) {defer wg.Done()var start, end int64for {select {case <-ctx.Done():// 如果接收到取消信号，退出协程returndefault:// --- 使用全局锁// 查找区间//mu.Lock()//start = startLoc//end = start + offset//startLoc = end//mu.Unlock()//if start >= endLoc {// return//}//if end > endLoc {// end = endLoc//}// --- 使用atomic原子操作start = atomic.LoadInt64(&startLoc)end = start + offsetif end > endLoc {end = endLoc}if start >= endLoc {return}// 应该不会出现ABA问题if ok := atomic.CompareAndSwapInt64(&startLoc, start, end); ok == false {//time.Sleep(100)continue}for i := start; i < end; i++ {if nums[i] == target {// 使用 atomic 以确保线程安全atomic.StoreInt32(&valid, 1)return}}}}
}var valid int32func main() {sliceLen := int64(100000)// 创建一个背景上下文和一个取消功能ctx := context.Background()// 假设 ddl 是一个固定的截止时间ddl := time.Now().Add(10 * time.Second) // 假设 5 秒钟后超时newCtx, cancel := context.WithDeadline(ctx, ddl)// 创建一个较大的切片 nums 并初始化nums := make([]int, sliceLen)endLoc = sliceLen// 初始化切片为随机数据，例如从 1 到 100，值为42的即为目标for i := 0; i < len(nums); i++ {nums[i] = i}startTime := time.Now()// 使用 WaitGroup 来等待所有协程完成var wg sync.WaitGroupoffset = int64(sliceLen / 10)// 启动多个协程进行查找for i := 0; i < 10; i++ {wg.Add(1)go find(nums, newCtx, &wg)}// 等待结果go func() {wg.Wait()cancel() // 等待所有协程结束后，调用 cancel}()// 检查结果select {case <-newCtx.Done():if atomic.LoadInt32(&valid) == 1 {fmt.Println("Found target!")} else {fmt.Println("Timeout or not found.")}}duration := time.Since(startTime)fmt.Printf("程序运行时间: %s\n", duration)
}
2.Context部分
2.1 context是并发安全
创建的初始context有两种 TODO()和Background()，查看内部结构体， 实际都是emptyCtx。
Background()创建的上下文通常被认为整个请求的顶级 Context，而TODO()创建的通常被认为是暂时的、未确定的 Context。
func Background() Context {return backgroundCtx{}
}func TODO() Context {return todoCtx{}
}1. 传值Value
直接对父context进行包装，并不会修改父context
type valueCtx struct {Contextkey, val any
}func WithValue(parent Context, key, val any) Context {if parent == nil {panic("cannot create context from nil parent")}if key == nil {panic("nil key")}if !reflectlite.TypeOf(key).Comparable() {panic("key is not comparable")}return &valueCtx{parent, key, val}
}2. 设置超时时间 WithDeadline
func WithDeadline(parent Context, d time.Time) (Context, CancelFunc) {return WithDeadlineCause(parent, d, nil)
}
2.2 context的信号传递
以cancel部分举例说明
1. 设置超时时间
设置取消函数的接口主要分为下列几种情况：
- 父Ctx为nil， 抛出异常
- 父Ctx具有超时时间，且比设置的超时时间更早结束，则新建CancelCtx加入父Ctx监听列表，且返回该新建CancelCtx。
- 设置新的包含超时时间的timerCtx（内部继承了cancelCtx结构体），加入父Ctx的监听列表，检查是否已经超时， 超时则取消该上下文， 没超时则设置计时器，等待取消。
func WithDeadlineCause(parent Context, d time.Time, cause error) (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{deadline: d,}c.cancelCtx.propagateCancel(parent, c)dur := time.Until(d)if dur <= 0 {c.cancel(true, DeadlineExceeded, cause) // deadline has already passedreturn c, func() { c.cancel(false, Canceled, nil) }}c.mu.Lock()defer c.mu.Unlock()if c.err == nil {c.timer = time.AfterFunc(dur, func() {c.cancel(true, DeadlineExceeded, cause)})}return c, func() { c.cancel(true, Canceled, nil) }
}2.设置子Ctx监听父Ctx
上下文取消传播：propagateCancel 的核心目的是将父上下文的取消信号（及其取消原因）传递给子上下文。不同的父上下文类型（如 *cancelCtx 或实现了 AfterFunc 方法的上下文）会采取不同的处理方式。
并发处理：通过 goroutines.Add(1) 和新的 goroutine 来监听父上下文的取消事件，确保并发场景下的取消传播。
其中分为三种情况：
- 父Ctx未设置Done ，则无需监听
- 父Ctx设置了回调函数
- 父Ctx类型是*cancelCtx，则把子Ctx加入自身map中，每个子Ctx都会开启协程监听父Ctx信号，同步取消自身。
主要就是依赖Channel进行信号传递。
func (c *cancelCtx) propagateCancel(parent Context, child canceler) {c.Context = parentdone := parent.Done()if done == nil {return // parent is never canceled}select {case <-done:// parent is already canceledchild.cancel(false, parent.Err(), Cause(parent))returndefault:}if p, ok := parentCancelCtx(parent); ok {// parent is a *cancelCtx, or derives from one.p.mu.Lock()if p.err != nil {// parent has already been canceledchild.cancel(false, p.err, p.cause)} else {if p.children == nil {p.children = make(map[canceler]struct{})}p.children[child] = struct{}{}}p.mu.Unlock()return}if a, ok := parent.(afterFuncer); ok {// parent implements an AfterFunc method.c.mu.Lock()stop := a.AfterFunc(func() {child.cancel(false, parent.Err(), Cause(parent))})c.Context = stopCtx{Context: parent,stop: stop,}c.mu.Unlock()return}goroutines.Add(1)go func() {select {case <-parent.Done():child.cancel(false, parent.Err(), Cause(parent))case <-child.Done():}}()
}参考链接：
Go 语言并发编程与 Context | Go 语言设计与实现
3.channel部分
3.1channel底层结构
在有缓冲区的channel部分，数据使用环形链表进行存储，存储有变量记录有效数据区域。
type hchan struct {qcount uint // Channel 中的元素个数dataqsiz uint // Channel 中的循环队列的长度buf unsafe.Pointer // Channel 的缓冲区数据指针elemsize uint16closed uint32elemtype *_type // element typesendx uint // Channel 的发送操作处理到的位置recvx uint // Channel 的接收操作处理到的位置recvq waitq // 等待消息的双向链表sendq waitq // 发生消息双向链表// lock protects all fields in hchan, as well as several// fields in sudogs blocked on this channel.// Do not change another G's status while holding this lock// (in particular, do not ready a G), as this can deadlock// with stack shrinking.lock mutex
}// 创建双向链表 构造等待消息 或 发生消息的goroutine的双向链表
type waitq struct {first *sudog last *sudog
}
有缓冲区
无缓冲区
3.2 对于不同的channel进行读入读出的不同情况
如果给一个 nil 的 channel 发送数据，会造成永远阻塞。
如果从一个 nil 的 channel 中接收数据，也会造成永久阻塞。
给一个已经关闭的 channel 发送数据， 会引起 panic。
从一个已经关闭的 channel 接收数据， 如果缓冲区中为空，则返回一个零值。
同时分为有缓冲区和无缓冲区两种，前者是异步的，在缓冲区未满时，可以持续输入，不会阻塞，直到缓冲区满；后者则为有goroutine输入，等待有协程进行数据消费，否则持续阻塞。
对nil的channel不可操作。
参考链接：
https://www.cnblogs.com/Paul-watermelon/articles/17484439.html
Go 语言 Channel 实现原理精要 | Go 语言设计与实现 (draveness.me)