Go——并发范式（每个请求一个goroutine）

绝地灬酷狼 2023-10-09 21:14 29阅读 0赞

### 每个请求一个goroutine ###

这种并发模式相对比较简单，就是来一个请求或任务就启动一个goroutine去处理，典型的就是Go中的HTTP server服务。下面看一下Go语言http标准库处理请求的方式，代码如下。

//net/http/server.go
    //func(srv *Server) Server(l net.Listener) 部分代码
    
    for {
        
    		//监听获取连接
    		rw, e := l.Accept()
    		if e != nil {
        
    			select {
        
    			case <-srv.getDoneChan():
    				return ErrServerClosed
    			default:
    			}
    			if ne, ok := e.(net.Error); ok && ne.Temporary() {
        
    				if tempDelay == 0 {
        
    					tempDelay = 5 * time.Millisecond
    				} else {
        
    					tempDelay *= 2
    				}
    				if max := 1 * time.Second; tempDelay > max {
        
    					tempDelay = max
    				}
    				srv.logf("http: Accept error: %v; retrying in %v", e, tempDelay)
    				time.Sleep(tempDelay)
    				continue
    			}
    			return e
    		}
    		tempDelay = 0
    		c := srv.newConn(rw)
    		c.setState(c.rwc, StateNew) // before Serve can return
    		//启动一个独立的goroutine处理该Web请求
    		go c.serve(ctx)
    	}

下面以计算100个自然数的和来举例，将计算任务拆分为多个task，每个task启动一个goroutine进行处理，示例代码如下：

package main
    
    import (
    	"fmt"
    	"sync"
    )
    
    func main() {
        
    	//创建任务通道
    	taskchan := make(chan task, 10)
    
    	//创建结果通道
    	resultchan := make(chan int, 10)
    
    	//wait用于同步等待任务的执行
    	wait := &sync.WaitGroup{
        }
    
    	//初始化task的goroutine，计算100个自然数之和
    	go InitTask(taskchan, resultchan, 100)
    
    	//每个task启动一个goroutine进行处理
    	go DistributeTask(taskchan, wait, resultchan)
    
    	//通过结果通道获取结果并汇总
    	sum := ProcessResult(resultchan)
    
    	fmt.Println("sum=", sum)
    }
    
    //工作任务
    type task struct {
        
    	begin  int
    	end    int
    	result chan<- int
    }
    
    //任务执行：计算begin到end的和
    //执行结果写入结果chan result
    func (t *task) do() {
        
    	sum := 0
    	for i := t.begin; i <= t.end; i++ {
        
    		sum += i
    	}
    	t.result <- sum
    }
    
    //构建task并写入task通道
    func InitTask(taskchan chan<- task, r chan int, p int) {
        
    	qu := p / 10
    	mod := p % 10
    	high := qu * 10
    	for j := 0; j < qu; j++ {
        
    		b := 10*j + 1
    		e := 10 * (j + 1)
    		tsk := task{
        
    			begin:  b,
    			end:    e,
    			result: r,
    		}
    		taskchan <- tsk
    	}
    	if mod != 0 {
        
    		tsk := task{
        
    			begin:  high + 1,
    			end:    p,
    			result: r,
    		}
    		taskchan <- tsk
    	}
    	close(taskchan)
    }
    
    //读取task chan，每个task启动一个worker goroutine进行处理
    //并等待每个task运行完，关闭结果通道
    func DistributeTask(taskchan <-chan task, wait *sync.WaitGroup, result chan int) {
        
    	for v := range taskchan {
        
    		wait.Add(1)
    		go ProcessTask(v, wait)
    	}
    	wait.Wait()
    	close(result)
    }
    
    //goroutine处理具体工作，并将处理结果发送到结果通道
    func ProcessTask(t task, wait *sync.WaitGroup) {
        
    	t.do()
    	wait.Done()
    }
    
    //读取结果通道，汇总结果
    func ProcessResult(resultchan chan int) int {
        
    	sum := 0
    	for r := range resultchan {
        
    		sum += r
    	}
    	return sum
    }

程序的逻辑分析：

1.  InitTask函数构建task并发送到task通道中。
2.  分发任务函数DistributeTask为每个task启动一个goroutine处理任务，等待其处理完成，然后关闭结果通道。
3.  ProcessResult函数读取并统计所有的结果。

这几个函数分别在不同的goroutine中运行，它们通过通道和sync.WaitGroup进行通信和同步。  
整个流程如图：  
![在这里插入图片描述][4996e7afa6a64bdc8ba9f582c694d09d.png]

[4996e7afa6a64bdc8ba9f582c694d09d.png]: https://img-blog.csdnimg.cn/4996e7afa6a64bdc8ba9f582c694d09d.png