新乡正规网站建设哪家便宜,化妆品网站建设描述,wordpress文章存在哪个数据库中,东莞网站建设推广方案前言
在实习的项目中有一个地方遇到了需要协程池的地方#xff0c;在mt推荐下使用了ants库。因此在此篇记录一下自己学习使用此库的情况。
场景描述
此服务大致是一个kafka消息接收、发送相关。接收消息#xff0c;根据参数设置定时器进行重发。 通过这里新建kafka服务在mt推荐下使用了ants库。因此在此篇记录一下自己学习使用此库的情况。
场景描述
此服务大致是一个kafka消息接收、发送相关。接收消息根据参数设置定时器进行重发。 通过这里新建kafka服务并在kratos框架的依赖注入使用。
func NewxxxxKafka {
RegisterSubscriber(context.TODO(), t, c.Group, false, handler.HandlerFunc(), nil)
}
这是其中的HandlerFunc在其中调用到Save部分逻辑操作。 func (h *xxxxxxxxxxx) HandlerFunc() broker.Handler {return func(ctx context.Context, event broker.Event) error {return h.uc.Save(ctx, msg)}
}
这个是我的逻辑部分其中RetryAtTime中会设置定时器并进行执行。在此之前涉及部分数据库操作。 func (uc *xxxx) Save() error {//部分数据库操作go uc.ms.RetryAtTime(ctx, id, UnixNextRetryTime)//其中有定时器
}
如果此处不使用协程那么所有消息接收都会因为定时器而卡死在主线程。
单使用wg.group的方案
单给定时器加协程 func (uc *xxxx) Save() error {var wg sync.WaitGroup//部分数据库操作go uc.ms.RetryAtTime(ctx, id, UnixNextRetryTime)//其中有定时器wg.Add(1)defer wg.Done()return nil
}
单给定时器使用协程并没有实现真正的并发只是在协程中去运行了定时器的逻辑其他所有操作依旧是单线程的。 这样可以解决定时器阻塞的问题但系统性能并没有提升。
消费消息时候加协程
将加协程移到HandlerFunc处可以实现并发的处理消息。 func (h *xxxxx) HandlerFunc() broker.Handler {return func(ctx context.Context, event broker.Event) error {h.wg.Add(1) // 增加等待组计数go func() {defer h.wg.Done() // 完成后减少等待组计数err : h.uc.Save(ctx, msg); err ! nil }()return nil}
}
这里需要注意不能在handlerfunc中去var wg sync.waitgroup。因为这样的话每次调用都会新声明一个那么每一个wg也只和一个goroutine关联了。需要写到结构体中。
但毕竟涉及到数据库操作协程中操作数据库可能会导致一些问题。虽然我这里业务逻辑好像只有一个增和查没什么影响
使用ants协程池
使用协程池管理协程:struct中新增pool
type xxxx struct {
// xxxxxxxxxxxxx//wg sync.WaitGroupPool *ants.Pool
}初始化在此处设置协程池的容量。因为考虑到有大量的定时器所以选择了一个相较于目前数据,较大的协程池。 func Newxxxx(xxxxx
) *xxxx {pool, err : ants.NewPool(10000)if err ! nil {log.Fatalf(failed to create ants Pool: %v, err)}return xxxx{//xxxxxPool: pool,}
}使用,主要是pool.Submit(func())函数,像池子中添加一个用于并发执行的函数即可。其余就交给池子底层去解决了。 func (h *xxx) HandlerFunc() broker.Handler {return func(ctx context.Context, event broker.Event) error {return h.Pool.Submit(func() {//xxxxxxif err : h.uc.Save(ctx, msg); err ! nil {h.logger.Error(failed to save message, error, err)}})}
}
记得在外层关闭池子 func NewxxxxxKafka() {for _, t : range c.Topics {if err : ks.RegisterSubscriber(context.TODO(), t, c.Group, false, handler.HandlerFunc(), nil); err ! nil {log.Fatal(err)}defer handler.Pool.Release()}}
总结
主要是记录了一次自己对协程池的使用在此过程中从仅使用协程处理定时器到使用协程处理整个方法实现并发再到使用协程池。后续会进行ants库底层源码的学习。
