go中关于chan应用的程序分析

程序员文章站 2024-03-22 08:32:46

...

文章目录

基础知识
前言
单中间者
多中间者

流程图
流程图分析
流程图总结

整体总结
之前的问题是否已经迎刃而解

基础知识

chan基本知识
go中关于for循环的知识
多进程通信

前言

学习go语言已经有几个月，但是关于go中的特性chan和routine的应用还不是很理解，如果搞不懂chan和routine的机制就很难流畅的用go编写出健壮的程序。所以我觉得关于chan应用的编程，是可以讲一讲的。
那我们首先会想到几个问题：

使用chan的代码与普通的代码有什么不同吗？
使用了chan后有什么好处吗？
怎么才能正确的使用chan呢？

我们先看第一个问题，众所周知 在go中chan是用来多个协程之间进行通信的，chan的应用思维是一种类似与client/servier的思维。也就是要有信息的生产者和消费者。既然如此，那就会设计到 不通协程之间的通信，而在顺序编程中是不会设计到通信的，是一种线性的流程。类似与下图：
go中关于chan应用的程序分析
在图中，我们看到，生产者和消费者之间的程序通过通信来进行相互影响的，而线性的程序是不会有这种问题，不管调用了多少函数，都只会在一条进程上顺序进行。
在下文中我们会用到一个中间者的概念。中间者是承接上游的消费者和开启下游的生产者。

单中间者

单中间者，就是在程序中只有一个线程连接生产者和消费者，不会进行扩增。

如下图所示：
go中关于chan应用的程序分析
在这种方式中，只有一个中间者，或者没有中间者。

gen.go

func gen(nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		for _, n := range nums {
			out <- n
		}
		close(out)
	}()
	return out
}

分析：创建一个out的通道，并打开一条协程将nums通过循环传递给通道out，在所有的nums都传入out后关闭通道。在打开协程后返回创建的out。
这个函数在程序中起着生产者的作用，生产者和消费这之间的交流通道就是这个out，当out中消费一个数后，本函数就会再将一个数推送到通道out中。

sq.go

func sq(in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		for n := range in {
			out <- n * n
		}
		close(out)
	}()
	return out
}

创建了通道out作为生产者的信息输出，并接受in作为消费者的信息输入。
这程序中，即是消费者也是生产者，并在中间进行了数据的处理nn。作为消费者，当通道in中可以取数据的时候，取出n并进行运算nn，然后作为生产者将结果推送到out中。当in中有值时会进行消费，并计算nn,当out为空时会进行生产，将nn的结果推到out中。

main.go

func main() {
	in := gen(2, 3)
	c1 := sq(in)
	for n := range c1 {
		fmt.Println(n) // 4  9
	}
}

main函数先调用了上面两个函数，然后用了一个循环进行输出。
输出这个循环就是整个程序的消费者，当c1中有值时，就会取出来放到n中。

单中间者并不能体现使用chan的优越性，整个过程仍然是类似于线性的流程进行。在下文中我们会开启多个中间者进行复杂的数据运算和处理，因为我们可以同时处理多条数据，必然给我们带来性能上的提高。

多中间者

单中间者，就是在程序中只有一个线程连接生产者和消费者，不会进行扩增。

go中关于chan应用的程序分析
在上图中，一个生产者被多个中间者处理生产的信息，以提高消费的效率。然后用一个merge的中间者收集各个中间者生产的信息，并将这些信息统一的交给消费者。

merge.go

func merge(cs ...<-chan int) <-chan int {
	var wg sync.WaitGroup
	out := make(chan int)

	// Start an output goroutine for each input channel in cs.  output
	// copies values from c to out until c is closed, then calls wg.Done.
	output := func(c <-chan int) {
		for n := range c {
			out <- n
		}
		wg.Done()
	}
	wg.Add(len(cs))
	for _, c := range cs {
		go output(c)
	}

	// Start a goroutine to close out once all the output goroutines are
	// done.  This must start after the wg.Add call.
	go func() {
		time.Sleep(222 * time.Second)
		wg.Wait()
		close(out)
	}()
	return out
}

这个merge函数给我带来了很大的困扰。

程序中有两个for循环，其中对cs的循环是针对数组的，循环的作用是对每个通道打开一个协程。而协程中的循环是监控通道用的，当通道close时推出循环。

为什么wg.Wait要用一个新的协程来，如果不用新的协程而写在函数体里面不行吗？当然是不行的，这里要注意一个与普通程序的分别：在使用chan的程序中，主程序的作用类似与初始化service和client而不进行具体的运算，所以肯定不能在初始化的时候进行wait，因为这里还是要进行生产，如果进行wait就不会运行main中的print（也就是消费者），而通道内的数据不进行消费就会形成阻塞，导致程序无法运行。
*那我已经知道了不能用在函数体里面，为什么我另起一条协程，不会出问题呢？？ 这涉及到后面部分，如果你明白了程序运行的整个流程，肯定可以解决这个疑问，我们把这个问题留到后面解决

main.go

func main() {
	in := gen(2, 3)
	// Distribute the sq work across three goroutines that both read from in.
	c1 := sq(in)
	c2 := sq(in)
	// Consume the merged output from c1 and c2.
	for n := range merge(c1, c2) {
		fmt.Println(n) // 4 then 9, or 9 then 4
	}
}

在函数中先用生产者，通过通道传递两个数，然后用两个sq中间者进行中间计算，然后用merge进行集合，最后通过print进行消费。

在初次看到这个程序的时候，我是觉得程序不会输出所有的结果的看，为什么我会这么看呢？因为在merge中的wait是在另起的一个协程中运行的，而不是在主程序中。如果不能理清楚整个程序的流程就会第一时间出现这种误解。那我们只有模拟程序一步一步的运行，来判断为什么这个程序可以顺利的运行。

流程图

因此我画了下面的流程图，图中标注了程序的step。（不通协程之间的step是可以重合的）

go中关于chan应用的程序分析
我们来过一遍上图的整个流程，你就发现程序中的所有疑问都迎刃而解。（虽然我们这里使用了多个中间者的方式，但是因为我们仅仅模拟了一个数据输入，所以只展示一个中间者就可以表示整个流程）

流程图分析

step 0：
[main] 调用程序 gen(3)

step 1：
[gen] 定义通道 out

step 2：

step 3：
[gen] 进入协程sub(gen)中的循环，判断循环是否结束，同时return out

step 4：
[main] 进入程序sq
[sq] 定义通道out，gen中的out定义为in
[sub(gen)] n-> out ##因为还没有遇到消费者，所以造成了阻塞

step 5：
[sq]进入协程sub(sq)中的循环，判断循环是否结束，同时return out

step 6：
[main] 进入程序merge
[merge] wg.Add(1)并判断cs[]循环时候结束
[sub(sq)] 从in中取出数n，并计算n*n,然后传入通道out ##这时in通道不阻塞 out 通道阻塞

step 7：
[sub(gen)] out不再阻塞，判断循环是否结束
[merge] 判断循环是否结束，如果没结束，进入sub(merge),结束进入sub(wait)

step 8：
[sub(gen)] 判断循环结束，close(out)
[sub(merge)]判断传入的通道是否关闭，
[sub(wait)] 判断merge协程是否都已经运行结束,没有结束，等待
[merge]return out 返回main

step 9：
[sub(merge)] 判断通道c未关闭，将c->out， ##这里out中的值没有消费，阻塞。c被消费不在阻塞
[main] 判断merge返回的通道t是否关闭

step10：
[sub(sq)] out通道开发，判断通道in是否关闭
[main] 通道t没有关闭，并输出t，此时t通道不在阻塞

step11：
[sub(sq)]in通道关闭（step8）,close(out)，协程结束
[main] t中没有值，不进行操作
[sub(merge)]out通道不再阻塞，判断c通道是否关闭

step12：
[sub(merge)] c通道关闭（step11） close(out) 协程结束

step13：
[sub(wait)]因为step12中协程结束，所以close(out)

step14：
[main]判断通道t关闭（step13），结束程序

流程图总结

只有等待，阻塞后输入，通道空后读出才会造成协程跳过一些step。

main主线程因为要等到通道t关闭，也就是要等待sub(wait)结束，而sub(wait)要等到所有的协程结束，所以肯定不会发生主线程跑完但是协程还没有跑完的情况。

整体总结

使用go就不可避免的要用到协程和通道，耐心的分析是写好代码的基础。当不了解程序的运行过程时，我们需要尽可能清晰的整理程序运行的流程。
在理解了程序的运行原理之后，就可以根据实际情况分析在什么时候使用多个中间者，使用多少个中间者，怎么进行收束等。（下面会给出使用多中间者的一个稍微复杂一点的代码，这时我在工作中使用的一个go的脚本）

之前的问题是否已经迎刃而解

为什么整个程序在主程序中没有等待协程程序结束的类似wg.Wait的语句，仍然程序可以正确的输出？
为什么在merge中wg.Wait是需要另外起一个协程，而不是在函数的主程序中进行wait？
make(chan int)和make(chan int ,1)的区别，在本程序中能把所有的chan声明从前者换成后者吗？

package main

import (
	"encoding/json"
	"errors"
	"fmt"
	_ "github.com/go-sql-driver/mysql"
	"github.com/jinzhu/gorm"
	"io/ioutil"
	"net/http"
	"regexp"
	"strings"
	"sync"
)

func product(in map[string]*Item) <-chan map[string]*Item {
	out := make(chan map[string]*Item)
	go func() {
		for k, v := range in {
			out <- map[string]*Item{k: v}
		}
		close(out)
	}()
	return out
}
func middle(in <-chan map[string]*Item) <-chan *Response {
	out := make(chan *Response)
	go func() {
		for aMap := range in {
			res := &Response{}
			for k, v := range aMap {
				dbRes := getDBByNameRes(v.Third)
				wg := sync.WaitGroup{}
				for _, item := range dbRes {
					wg.Add(1)
					go func(item *CateGoryField) {
						defer func() {
							wg.Done()
							if err := recover(); err != nil {
								fmt.Printf("book, panic: %+v\n", err)
							}
						}()
						resultItem := &Result{}
						resultItem.CopyrightCateGory = k
						getResult(item, resultItem)
						if resultItem.First.ID == v.FirstID && resultItem.Second.Name == v.Name {
							responseItem := &Response{}
							responseItem.CopyrightCateGory = resultItem.CopyrightCateGory
							responseItem.First = resultItem.First.ID
							responseItem.Second = resultItem.Second.ID
							responseItem.Third = resultItem.Third.ID
							res = responseItem
						}
					}(item)
				}
				wg.Wait()
			}
			out <- res
		}
		close(out)
	}()
	return out
}
func mergee(in []<-chan *Response) <-chan *Response {
	out := make(chan *Response)
	wg := sync.WaitGroup{}

	output := func(c <-chan *Response) {
		for n := range c {
			out <- n
		}
		wg.Done()
	}
	for _, i := range in {
		wg.Add(1)
		go output(i)
	}
	go func() {
		wg.Wait()
		close(out)
	}()
	return out
}
func main() {
	phpMap, err := parseToMap()
	if err != nil {
		fmt.Println("解析成map错误")
		return
	}
	type resultItems map[string]Result
	response := make(map[string][]Response)

	s := product(phpMap)
	v := make([]<-chan *Response, len(phpMap))
	for i := 0; i < len(phpMap); i++ {
		v[i] = middle(s)
	}
	out := mergee(v)
	for item := range out {
		response["category_mappers"] = append(response["category_mappers"], *item)
	}

	for k := range phpMap {
		var yn bool
		for _, v := range response["category_mappers"] {
			if k == v.CopyrightCateGory {
				yn = true
				break
			}
		}
		if yn {
			continue
		}
		fmt.Println(k, phpMap[k], " not in the category_mappers")
	}
	resultJson, err := json.Marshal(response)
	//file, err := os.OpenFile("test.json",  os.O_RDWR | os.O_APPEND | os. O_CREATE,066)
	//defer file.Close()
	//n,err:=file.WriteString(string(resultJson))
	ioutil.WriteFile("test.json", resultJson, 0777)
	fmt.Println(string(resultJson), err)
}

var php = `
[
        '玄幻小说' => ['first_id' => 1, 'name' => '玄幻', 'third' => '东方玄幻'],
        '灵异悬疑' => ['first_id' => 1, 'name' => '灵异', 'third' => '灵异鬼怪'],
    ]
`

type Item struct {
	FirstID int    `json:"first_id"`
	Name    string `json:"name"`
	Third   string `json:"third"`
}
type CategoryReplaceMap struct {
	ID     int    `json:"id"`
	Before string `json:"before"`
	After  string `json:"after"`
}
type Result struct {
	CopyrightCateGory string         `json:"copyright_category"`
	First             *CateGoryField `json:"stack_first_category_id"`
	Second            *CateGoryField `json:"stack_second_category_id"`
	Third             *CateGoryField `json:"stack_third_category_id"`
}
type Response struct {
	CopyrightCateGory string `json:"copyright_category"`
	First             int    `json:"stack_first_category_id"`
	Second            int    `json:"stack_second_category_id"`
	Third             int    `json:"stack_third_category_id"`
}

func parseToMap() (map[string]*Item, error) {
	var phpMap = make(map[string]*Item)
	rePlaceMap, err := getReplaceMapFromDB()
	if err != nil {
		fmt.Println(err)
	}
	fmt.Println("分类改变的对应关系")
	for k, v := range rePlaceMap {
		fmt.Println(k, v)
	}
	phpjson := strings.ReplaceAll(php, " ", "")
	for k, v := range rePlaceMap {
		phpjson = strings.ReplaceAll(phpjson, k, v)
	}
	phpjson = strings.Replace(phpjson, "=>", ":", -1)
	phpjson = strings.Replace(phpjson, "'", "\"", -1)
	phpjson = strings.Replace(phpjson, "[", "{", -1)
	phpjson = strings.Replace(phpjson, "]", "}", -1)
	reg1 := regexp.MustCompile(`([\d]+) *:`)
	if reg1 == nil {
		fmt.Println("regexp err")
		return nil, errors.New("匹配模式错误1")
	}
	//根据规则提取关键信息
	phpjson = reg1.ReplaceAllString(phpjson, `"${1}" :`)
	reg2 := regexp.MustCompile(`(}),(\n *})`)
	if reg2 == nil {
		fmt.Println("regexp err")
		return nil, errors.New("匹配模式错误1")
	}
	//根据规则提取关键信息
	phpjson = reg2.ReplaceAllString(phpjson, `${1}${2}`)
	fmt.Println(phpjson)
	err = json.Unmarshal([]byte(phpjson), &phpMap)
	fmt.Println(err)
	return phpMap, err
}

func getReplaceMapFromDB() (map[string]string, error) {
	res := []*CategoryReplaceMap{}
	//res:= execsql(sql)
	db, err := gorm.Open("mysql", "browser:aaa@qq.com(rm-8vba1p18akz3f2j62.mysql.zhangbei.rds.aliyuncs.com:3306)/copyright_manage?charset=utf8&parseTime=True&loc=Local")
	if err != nil {
		fmt.Println("连接数据库失败")
		return nil, err
	}
	err = db.Table("category_change").Find(&res).Error
	if err != nil {
		fmt.Println("查询数据库失败")
		return nil, err
	}
	response := make(map[string]string)

	for _, item := range res {
		item.Before = strings.ReplaceAll(item.Before, "\n", "")
		item.After = strings.ReplaceAll(item.After, "\n", "")
		item.Before = strings.ReplaceAll(item.Before, " ", "")
		item.After = strings.ReplaceAll(item.After, " ", "")
		response[item.Before] = item.After
	}
	return response, nil
}

func getResult(item *CateGoryField, resultItem *Result) {
	if item.Rank == 1 {
		resultItem.First = item
	}
	if item.Rank == 2 {
		resultItem.Second = item
		dbRes := getDBByID(item.Parent)
		getResult(dbRes, resultItem)
	}
	if item.Rank == 3 {
		resultItem.Third = item
		dbRes := getDBByID(item.Parent)
		getResult(dbRes, resultItem)
	}
}

type CateGoryField struct {
	Name   string `json:"name"`
	Parent int    `json:"parent"`
	Rank   int    `json:"rank"`
	ID     int    `json:"id"`
}

func getDBByNameRes(name string) []*CateGoryField {
	res := make([]*CateGoryField, 0)
	//res:= execsql(sql)
	db, err := gorm.Open("mysql", "distribution_w:********@tcp(rm-8vba1p18akz3f2j62.mysql.zhangbei.rds.aliyuncs.com:3306)/content_distribution?charset=utf8mb4&parseTime=True&loc=Local")
	if err != nil {
		fmt.Println("连接数据库失败")
		return nil
	}
	err = db.Table("category").Where("name = ? ", name).Find(&res).Error
	if err != nil {
		fmt.Println("查询数据库失败")
		return nil
	}
	return res
}
func getDBByID(id int) *CateGoryField {
	res := &CateGoryField{}
	//res:= execsql(sql)
	db, err := gorm.Open("mysql", "distribution_w:*********@tcp(rm-8vba1p18akz3f2j62.mysql.zhangbei.rds.aliyuncs.com:3306)/content_distribution?charset=utf8mb4&parseTime=True&loc=Local")
	if err != nil {
		fmt.Println("连接数据库失败")
		return nil
	}
	err = db.Table("category").Where("id = ? ", id).Find(&res).Error
	if err != nil {
		fmt.Println("查询数据库失败")
		return nil
	}
	return res
}
func getReplaceMapFromKM() (map[string]string, error) {
	var rePlaceMap = make(map[string]string)
	client := http.Client{}
	request, err := http.NewRequest("GET", "https://km.qutoutiao.net/pages/viewpage.action?pageId=*****", nil)
	request.Header.Add("cookie", "JSESSIONID=

 - [ ] ******

")
	resp, err := client.Do(request)
	body, _ := ioutil.ReadAll(resp.Body)

	//查找table
	reg := regexp.MustCompile(`细节问题.*?(<tbody>.*?</tbody>)`)
	if reg == nil {
		fmt.Println("regexp err")
		return nil, errors.New("匹配模式错误1")
	}
	tables := reg.FindStringSubmatch(string(body))
	if tables == nil {
		return nil, errors.New("匹配table错误")
	}
	table := tables[1]
	table = strings.ReplaceAll(table, "&gt;", ">")
	table = strings.ReplaceAll(table, " ", "")

	//查找table中的tr
	reg = regexp.MustCompile(`<tr>(.*?)</tr>`)
	if reg == nil {
		fmt.Println("regexp err")
		return nil, errors.New("匹配模式错误2")
	}
	trs := reg.FindAllStringSubmatch(table, -1)
	if trs == nil {
		fmt.Println("regexp err")
		return nil, errors.New("匹配模式错误2")
	}
	//第一个为field标题
	for _, tr := range trs[1:] {
		fmt.Println("tr\n", tr[0], "\n")
		reg = regexp.MustCompile(`<td.*?>(.*?)</td>`)
		if reg == nil {
			fmt.Println("regexp err")
			return nil, errors.New("匹配模式错误2")
		}
		tds := reg.FindAllStringSubmatch(tr[0], -1)

		//fmt.Println("td\n", tr[0])
		reg = regexp.MustCompile(`(\[.*?\])`)
		if reg == nil {
			fmt.Println("regexp err")
			return nil, errors.New("匹配模式错误2")
		}
		beforeItem := reg.FindAllStringSubmatch(tds[0][1], -1)
		afterItem := reg.FindAllStringSubmatch(tds[1][1], -1)
		rePlaceMap[beforeItem[0][1]] = afterItem[0][1]
	}
	return rePlaceMap, err
}
**

相关标签： go chan 程序分析 channel

上一篇：记一次“Axis2客户端调用产生大量CLOSE_WAIT连接”的解决过程博客分类：原创 axis2webserviceGCclose_wait

下一篇：用代码判断当前系统是否支持某个版本的feature feature版本检查版本控制jdkjdk9

go中关于chan应用的程序分析

文章目录

基础知识

前言

单中间者

多中间者

流程图

流程图分析

流程图总结

整体总结

之前的问题是否已经迎刃而解

go中关于chan应用的程序分析

并行算法的性能度量——函数句柄在并行程序分析中的应用（2-4）

php程序中应用swfupload上传关于SESSION的问题

php程序中应用swfupload上传关于SESSION的问题

flutter应用程序开发中Contrainer 组件的基本使用以及宽度限制分析

Java中关于应用领域的分析

Java中关于应用领域的分析

php程序中应用swfupload下传关于SESSION的有关问题

Apache Kafka应用程序中的机器学习和实时分析（第二部分）

Apache Kafka应用程序中的机器学习和实时分析（第一部分）