如何使用Go创建工作池? 配合计时器使用工作池 图文讲解 [_]~(￣▽￣)~* Go相关

使用go程通信创建工作池

创建工作池会用到我上述的大部分特性

func 工作函数(id int, 任务 <-chan int, 结果 chan<- int) {
	for i := range 任务 {
		fmt.Printf("工作池:%d ++++++++ 任务:%d\n", id, i)
		time.Sleep(time.Second)
		fmt.Printf("工作池:%d -------- 任务:%d\n", id, i)
		结果 <- i * 2
	}
}

func main() {
	// 工作池
	任务 := make(chan int, 100)
	工作结果 := make(chan int, 100)
	// 创建工作池,开启5个go程
	for 工作池 := 1; 工作池 <= 5; 工作池++ {
		go 工作函数(工作池, 任务, 工作结果)
	}
	// 创建与分配任务
	for 创建任务 := 1; 创建任务 <= 5; 创建任务++ {
		任务 <- 创建任务
	}
	// 输出结果
	for 结果 := 1; 结果 <= 5; 结果++ {
		fmt.Println("工作结果:", <-工作结果)
	}
}

这里工作池和任务的数量都是可以更加需求动态改变的, 可以更加CPU的线程数创建对应数量的工作池(runtime.NumCPU返回当前CPU的线程数)

计时器(time.Timer)

网上已经有很多计时器相关的文章了, 如果你对计时器还不了解, 可以看一下这个大佬的文章一起学Go之计时器（Timer/Tick）, 用计时器的时候需要注意内存泄漏。
简单说: go计时器分为两种, 一种执行一次就结束的常规计时器(time.Timer), 一种是按照计时一直会不断重复执行的循环计时器(time.Ticker)。
其中time.Timer创建计时器的方法主要有两种

1. time.NewTimer(时间): 计时结束后会像内部的变量C(是一个时间类型管道)返回当前时间
2. time.AfterFunc(时间, 函数()) 计时结束后会执行函数()
   常规的计时器, 可以重计(计时器.Reset(时间))或者关闭计时器(计时器. Stop())
   而另一种循环计时器(time.Ticker)是无法进行重计的, 只能进行关闭操作。

var 计时管道 = make(chan int, 5)

func 倒计时(时间 chan int) {
	i := <-时间
	time.Sleep(time.Second)
	时间 <- i + 1
}

func main() {
	计时器 := time.NewTimer(time.Second * 5)
	计时管道 <- 0
	for {
		go 倒计时(计时管道)
		select {
		case 结束时间 := <-计时器.C:
			fmt.Println("结束了, ", 结束时间)
			goto 结束
		case 当前时间 := <-计时管道:
			计时管道 <- 当前时间
			fmt.Println(当前时间)
		}
	}
结束:
}

计时器主要使用的场景之一就是超时关闭, 而且往往会配合 select - case进行使用, 下面我就用工作池配合计时器来完善工作池的机制。

工作池与计时器

上述工作池还有一个很不完善的地方, 上述程序想要正常工作, 我们需要事先知道其创建了多个任务, 来决定输出结果的数量, 但实际环境中, 这样的理想条件一般不会出现, 而更加长间的是利用超时机制, 等待一段时间后来彻底结束, Go的标准库中自带的计时器即可完成此功能。

func 工作函数(id int, 任务 <-chan int, 结果 chan<- int) {
	for i := range 任务 {
		fmt.Printf("工作池:%d ++++++++ 任务:%d\n", id, i)
		time.Sleep(time.Second)
		fmt.Printf("工作池:%d -------- 任务:%d\n", id, i)
		结果 <- i * 2
	}
}

func main() {
	// 工作池
	任务 := make(chan int, 100)
	工作结果 := make(chan int, 100)
	// 创建工作池,开启5个go程
	for 工作池 := 1; 工作池 <= 3; 工作池++ {
		go 工作函数(工作池, 任务, 工作结果)
	}
	// 创建与分配任务
	for 创建任务 := 1; 创建任务 <= 5; 创建任务++ {
		任务 <- 创建任务
	}
	// 创建计时器, 设置超时时间
	计时器 := time.NewTimer(10 * time.Second)
	defer 计时器.Stop()
	for i := 0; i < 1; {
		select {
		case 结果 := <-工作结果:
			计时器.Reset(10 * time.Second)
			fmt.Printf("当前Go程数量: %v, 工作结果: %v \n", runtime.NumGoroutine(), 结果)
		case <-计时器.C:
			i += 1
			// 这里把不用的管道销毁一下, 减少Go程数量, 同时也能避免内存泄漏
			close(工作结果)
			close(任务)
			// 等待go的垃圾回收机制来回收一下没用的Go程
			time.Sleep(time.Second)
			fmt.Println("当前Go程数量:", runtime.NumGoroutine())
			break
		}
	}
}