for range 中使用闭包
先来看一个示例
package main
import (
"fmt"
"sync"
)
type T struct{
x int
}
func main(){
ch := make(chan int, 5)
var wg sync.WaitGroup
wg.Add(1)
go func(){
defer wg.Done()
for a:= range ch {
wg.Add(1)
go func(){
defer wg.Done()
fmt.Println(a)
}()
}
}()
for i:=0;i<10;i++ {
ch <- i
}
close(ch)
wg.Wait()
}
那么这段代码的输入结果是?
9
3
8
8
8
9
8
8
8
9
这和预期是不符的(预期随机输入1-9的数字),那原因是什么呢?
每次遍历都会讲值进行复制,为了避免过大的内存占用,会在循环的最开始创建一个实例,在每次的遍历过程中,将数据复制到这个实例中。而上面示例代码中的闭包函数仅会持有实例的引用(而不会复制数据),这样就导致所有的go routines
实质上拿到的是同一个实例的引用。
结论
因此要解决这个问题,可以在每次遍历时候创建一个新的实例,然后将值赋值给它:
func main(){
...
go func(){
defer wg.Done()
for a:= range ch {
wg.Add(1)
i := a
go func(){
defer wg.Done()
fmt.Println(i)
}()
}
}()
...
}
func main(){
...
go func(){
defer wg.Done()
for a:= range ch {
wg.Add(1)
i := a
go func(a int){
defer wg.Done()
fmt.Println(a)
}(a)
}
}()
...
}
:=的错误用法
示例
package main
import (
"fmt"
"os"
)
func main() {
var data []string
killswitch := os.Getenv("KILLSWITCH")
if killswitch == "" {
fmt.Println("kill switch is off")
data, err := getData()
if err != nil {
panic("ERROR!")
}
fmt.Printf("Data was fetched! %d\n", len(data))
}
for _, item := range data {
fmt.Println(item)
}
}
func getData() ([]string, error) {
// Simulating getting the data from a datasource - lets say a DB.
return []string{"there","are","no","strings","on","me"}, nil
}
输入结果:
kill switch is off
Data was fetched! 6
为什么对data的遍历输出没有生效?
原因是当我们使用:=
时,data
和err
都会被当做新的变量,换句话说,data
是在大括号范围内的新定义的局部变量,当代码执行到大括号结束时,data
和err
将被销毁
结论
当函数内部有多个函数块使用函数范围内定义的变量时,尽量不要使用:=
,而是提前声明好变量,在对变量赋值时使用=
将上面代码修改一下
func main() {
var data []string
var err error
killswitch := os.Getenv("KILLSWITCH")
if killswitch == "" {
fmt.Println("kill switch is off")
data, err = getData()
...
}
workderPool
示例
package main
import (
"fmt"
"sync"
"time"
)
type A struct{
id int
}
func main() {
start := time.Now()
channel := make(chan A, 100)
var wg sync.WaitGroup
wg.Add(1)
go func(){
defer wg.Done()
for a:= range channel{
wg.Add(1)
go func(a A){
defer wg.Done()
process(a)
}(a)
}
}()
for i:=0;i<10000000;i++{
channel <-A{i}
}
close(channel)
wg.Wait()
elasped := time.Since(start)
fmt.Printf("Took %s\n", elasped)
}
func process(a A){
fmt.Printf("Start process %v\n", a)
time.Sleep(100 * time.Millisecond)
fmt.Printf("Finish process %v\n", a)
}
如果上面代码遍历的次数非常大就会执行错误,原因是,当创建一个go routine
时都需要为其开辟一块内存空间,当其执行完成时回收。因此,创建越多的go routine
就会产生更多的内存占用,而且go routine
的执行是需要CPU进行调度的,如果cpu的内核越少,在内存中等待调度的go routine
就越多。那么如何解决这个问题呢?
答案显而易见,越多的go routine
可以带来越快的执行效率,但是也会占用大量的资源,我们需要在中间找一个平衡,或者说需要对go routine
的数量进行把控,将代码做些许修改,限制work pools
的数量为100:
func main() {
start := time.Now()
channel := make(chan A, 100)
var wg sync.WaitGroup
var workerPoolSize = 100
wg.Add(1)
go func(){
defer wg.Done()
for i:=0;i<workerPoolSize;i++{
wg.Add(1)
go func(){
defer wg.Done()
for a:= range channel{
process(a)
}
}()
}
}()
for i:=0;i<100000;i++{
channel <-A{i}
}
close(channel)
wg.Wait()
elasped := time.Since(start)
fmt.Printf("Took %s\n", elasped)
}
针对上面这段代码, 可以讲channel看做是队列,二每个worker则看做是一个consumer
,多个worker
监听同一个channel
。当然这只是实现worker pool
的简单示例,实现一个真正意义的worker pool
还需要做很多工作,比如动态伸缩、参数配置化等等
有疑问加站长微信联系(非本文作者)