go 协程池的实现

使用场景

这次需求是做一个临时的数据采集功能,为了将积压的数据快速的消耗完,但是单一的脚本消耗的太慢,于是乎就手写了一个简单的协程池:

  1. 为了能加快数据的收集速度
  2. 为了稳定协程的数量,让脚本变得稳定

设计图如下

在这里插入图片描述
协程池中提供了三个方法:

  1. 一个是Addjob用来将任务加入到任务池中
  2. Do 是用来消耗任务池中的任务
  3. HandleErrors 用来获取到错误信息
  4. Stop 是当脚本停止以后,不会立刻停止而是等待所有的人物消耗光在停止

代码如下

该协程池是借用了go扩展库中的semaphore来实现的。

  1. semaphore 信号量是一种同步机制,用于控制对共享资源的访问,常用于限制可以同时访问某一资源或资源池的线程数量。
  2. 我使用的是Acquire函数来实现的,Acquire 当资源访问量达到上限时会被阻塞,直到有协程执行完成,所以我们这里需要对Acquire的上下文设置超时时间,防止我们的任务出现死任务无法退出,从而导致整个协程池堵死。
  3. 我们在任务执行完成后要通过Release来释放资源,防止我们池子越变越小。
package pool

import (
	"context"
	"sync"
	"time"

	"golang.org/x/sync/semaphore"
)

type GoPool struct {
	MaxNum int
	Jobs   chan func() error
	sem    *semaphore.Weighted
	wg     *sync.WaitGroup
	Errs   chan error
}

func NewGoPool(num int) *GoPool {
	return &GoPool{
		MaxNum: num,
		Jobs:   make(chan func() error, num),
		sem:    semaphore.NewWeighted(int64(num)),
		wg:     &sync.WaitGroup{},
		Errs:   make(chan error, num),
	}
}

func (g *GoPool) Do() {

	go g.gAcquire()
}

func (g *GoPool) AddJob(f func() error) {
	g.Jobs <- f
}

func (g *GoPool) gAcquire() {

	for {
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		select {
		case job, ok := <-g.Jobs:
			if !ok {
				cancel()
				return
			}
			g.wg.Add(1)
			if err := g.sem.Acquire(ctx, 1); err != nil {
				// g.Errs <- err
				g.wg.Done()
				cancel() // 确保在退出前取消context
				break
			}
			go func() {
				defer g.sem.Release(1)
				defer g.wg.Done()
				if err := job(); err != nil {
					g.Errs <- err
					return
				}
			}()
		case <-ctx.Done():
			return
		default:
			continue
		}
	}
}

func (g *GoPool) Stop() {
	close(g.Jobs)
	g.wg.Wait()
	close(g.Errs)
}


func (g *GoPool) HandleErrors(handler func(error)) {
	for err := range g.Errs {
		handler(err)
	}
}

相关推荐

  1. go实现

    2024-07-23 10:36:04       55 阅读
  2. Go实现简单(通过channel实现

    2024-07-23 10:36:04       43 阅读
  3. golang 实现

    2024-07-23 10:36:04       69 阅读
  4. go

    2024-07-23 10:36:04       39 阅读
  5. 面:go能不能手写一个简单

    2024-07-23 10:36:04       46 阅读
  6. 简单剖析tRPC-Go中使用第三方ants

    2024-07-23 10:36:04       47 阅读
  7. Lua

    2024-07-23 10:36:04       46 阅读
  8. Go 通过 goroutines 实现类似线模式

    2024-07-23 10:36:04       54 阅读

最近更新

  1. docker php8.1+nginx base 镜像 dockerfile 配置

    2024-07-23 10:36:04       142 阅读
  2. Could not load dynamic library ‘cudart64_100.dll‘

    2024-07-23 10:36:04       156 阅读
  3. 在Django里面运行非项目文件

    2024-07-23 10:36:04       131 阅读
  4. Python语言-面向对象

    2024-07-23 10:36:04       141 阅读

热门阅读

  1. Python语言-面向对象

    2024-07-23 10:36:04       141 阅读
  2. 开封建筑设计资质申请正确填写信息

    2024-07-23 10:36:04       82 阅读
  3. Android中接入hook框架:lancet-base

    2024-07-23 10:36:04       71 阅读
  4. 如何平衡硬约束与软约束

    2024-07-23 10:36:04       75 阅读
  5. 医学深度学习与机器学习融合的随想

    2024-07-23 10:36:04       76 阅读
  6. 19、基于共享内存的内存池

    2024-07-23 10:36:04       65 阅读
  7. Hisilicon适配按键

    2024-07-23 10:36:04       64 阅读
  8. qml实现双指左右滑动,上下滑动

    2024-07-23 10:36:04       47 阅读