动态创建goroutine和channel

Dear 丶 2022-12-28 12:47 100阅读 0赞

若各位看官有其它较好的思想和方式，欢迎留言共同交流。

动态的创建需求一般应用于动态的数据量、动态的时间等，比如这里遇到的可能是数十万、百万左右的数据量推送。可以采取的方式比如按数据量来判断，以十万数据量为一个小单位，十万数据量对应一个channel、一个读goroutine以及一个写goroutine，因为每次推送的目标数量不定，因此采取动态创建的方式也可达到并发执行目的，整体goroutine的数量也是可控的。

其它能想到的业务应用背景如：每天的数据在第二天凌晨进行加载同步，数据量都很大，那么可以根据数据量来决定goroutine和ch的数量；也可根据时间点来决定如何创建goroutine，在我们肯定业务背景的情况下，该业务每个小时都会产生大量数据存放于A，那么我们第二天凌晨需要同步A昨天一整天的数据到B，按时间点来做可以做成一个小时的数据由2个goroutine、一个ch来完成，这样整个过程也只需48个goroutine就可以完成。

以下来源于不断实践(代码只展示整体概况哦)，一步步来看看。

## concurrency one ##

不论有多少数据量，基于三写三读来完成，num为条数总数；

go func() {
    		for i := 1; i <= num; i++ {
    			if (i-1)%3 == 0 {
    				reqCh1 <- Req{Bs: bs, Index: i}
    			}
    		}
    		close(reqCh1)
    	}()
    
    	go func() {
    		for i := 1; i <= num; i++ {
    			if (i-2)%3 == 0 {
    				reqCh2 <- Req{Bs: bs, Index: i}
    			}
    		}
    		close(reqCh2)
    	}()
    
    	go func() {
    		for i := 1; i <= num; i++ {
    			if (i-3)%3 == 0 {
    				reqCh3 <- Req{Bs: bs, Index: i}
    			}
    		}
    		close(reqCh3)
    	}()
       
        go func() {
    		for input := range reqCh3 {
    			// do req
    		}
    		resMap["A"] = struct{}{}
    		fmt.Println("reqCh3推送结束.A=", resMap["A"])
    	}()
    
    	go func() {
    		for input := range reqCh2 {
                // do req
    		}
    		resMap["B"] = struct{}{}
    		fmt.Println("reqCh2推送结束.B=", resMap["B"])
    	}()
    	go func() {
    		for input := range reqCh1 {
    			// do req
    		}
    		resMap["C"] = struct{}{}
    		fmt.Println("reqCh1推送结束.C=", resMap["C"])
    	}()
    
    	go func() {
    	over:
    		for {
    			if len(resMap) == 3 {
    				for k := range resMap {
    					fmt.Println("-------->推送结束，打印key==", k)
    				}
    				close(signCh)
    			}
    			select {
    			case <-signCh:
    				close(notifyCh)
    				break over
    			default:
    
    			}
    		}
    	}()
    
    	<-notifyCh

此种方式基本通用，但数据量过大时也会比较慢；

## concurrency two ##

通用化实现。按实际数据量大小动态判断，基于数据量大小动态创建goroutine和ch；

chMap, intervalMap := make(map[string]chan Req), make(map[int]TimeStartAndEnd)	
    count := num / 100000    // 以十万数据为一个单位
    	if num%100000 != 0 {
    		count = count + 1
    		fmt.Println(fmt.Sprintf("num=%v,对count执行了+1.count=%v", num, count))
    	}
     
        for i := 1; i <= count; i++ {
    		chMap["ch-"+strconv.Itoa(i)] = make(chan Req)
    		start, end := 1, 0
    		if i != 1 {
    			start = (i - 1) * 100000
    		}
    		if i == count {
    			end = 100000*i + 1
    		} else {
    			end = 100000 * i
    		}
    		// 以十万条为一个单位,不够十万的当作一个单位 ;第1个=[1,100000),第2个=[100000,200000),第3个=[200000,300000)
    		intervalMap[i] = TimeStartAndEnd{Index: i, Start: start, End: end}
    	}
    
        for k, v := range chMap {
    		names := strings.Split(k, "-")  
    		sighInt, _ := strconv.Atoi(names[1]) // 标号，1,2,3...
    		interval := intervalMap[sighInt]     
    		go func(name string, ch chan Req, start, end, id int) {
    			fmt.Println(fmt.Sprintf("协程%v=%v的执行区间为[%v,%v]", id, name, start, end))
    			for i := start; i < end; i++ {
                    // do sth
    				ch <- Req{Bs: bs, Index: i}
    			}
    			close(ch)
                return
    		}(k, v, interval.Start, interval.End, sighInt)
    	}
        
        resMap := make(map[string]struct{})
    	for k, v := range chMap {
    		names := strings.Split(k, "-")  
    		sighInt, _ := strconv.Atoi(names[1]) // 标号，1,2,3...
    		go func(name string, ch chan Req, sighInt int) {
    			for rsp := range ch {
                    // do sth
    				fmt.Println("协程", name, "正在处理...")
    			}
    			resMap[name] = struct{}{}
    			fmt.Println(fmt.Sprintf("协程%v读取完毕.", name))
                return
    		}(k, v, sighInt)
    	}
    
        signCh, notifyCh := make(chan struct{}), make(chan struct{})
    	go func() {
    	over:
    		for {
    			if len(resMap) == count {
    				for k := range resMap {
    					fmt.Println(fmt.Sprintf("-------->推送结束，打印key=%v", k))
    				}
    				close(signCh)
    			}
    			select {
    			case <-signCh:
    				close(notifyCh)
    				break over
    			default:
    
    			}
    		}
    	}()
    
    	<-notifyCh

当然也可以加入errCh来流入错误信息等等额外处理。该模型也可以应对大部分情景。

## concurrency three ##

待补充优化...

**go并发之美·干货好文合集：**

[go并发之美·redis篇·接口幂等性设计实现][go_redis]

[go并发之美·redis篇·实现实时排行榜][go_redis 1]

[go并发日记·多goroutine并发安全的操作共享变量][go_goroutine]

[go并发日记·常用并发场景中的并发控制思路与方案][go]

[go并发日记·避免goroutine泄漏/实现协程可控][go_goroutine 1]

[go并发日记·深入理解go语言channel与goroutine阻塞][go_go_channel_goroutine]

[go并发日记·基于channel优雅的进行并发控制][go_channel]

[go并发日记·WaitGroup并发控制探索及实际场景运用][go_WaitGroup]

[channel死锁的情况分析][channel]

[go并发之美·B操作伴随A操作(B和A同步)][go_B_A_B_A]

[go多函数并发通用工具的设计与实现][go 1]

[go_redis]: https://lan6193.blog.csdn.net/article/details/108938119
[go_redis 1]: https://lan6193.blog.csdn.net/article/details/108931188
[go_goroutine]: https://lan6193.blog.csdn.net/article/details/105715373
[go]: https://lan6193.blog.csdn.net/article/details/108670428
[go_goroutine 1]: https://lan6193.blog.csdn.net/article/details/105559568
[go_go_channel_goroutine]: https://lan6193.blog.csdn.net/article/details/97528721
[go_channel]: https://lan6193.blog.csdn.net/article/details/101201408
[go_WaitGroup]: https://lan6193.blog.csdn.net/article/details/100887891
[channel]: https://lan6193.blog.csdn.net/article/details/97656420
[go_B_A_B_A]: https://lan6193.blog.csdn.net/article/details/108901339
[go 1]: https://lan6193.blog.csdn.net/article/details/103298105