经典的企业网站,外链提交网站,显示网站正在建设中,网页图片提取器题目
手撕 对无序的切片查询指定数 使用context进行子协程的销毁 并且进行超时处理。
全局变量定义
var (startLoc int64(0) // --- 未处理切片数据起始位置endLoc int64(0) // --- 切片数据右边界 避免越界offset int64(0) // --- 根据切片和协程数量 在主线程 动态设…题目
手撕 对无序的切片查询指定数 使用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 (contextfmtsyncsync/atomictime
)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, startLocoffset)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 (contextfmtsyncsync/atomictime
)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 Gs 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)
