前言

在实习的项目中有一个地方遇到了需要协程池的地方，在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.WaitGroup
	Pool *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{
	//xxxxx
		Pool:   pool,
	}
}

使用,主要是pool.Submit(func())函数,像池子中添加一个用于并发执行的函数即可。其余就交给池子底层去解决了。

func (h *xxx) HandlerFunc() broker.Handler {
	return func(ctx context.Context, event broker.Event) error {
		return h.Pool.Submit(func() {
			//xxxxxx
			if 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库底层源码的学习。