Go高级特性探究之处理1分钟百万请求详解

在处理1分钟百万请求时，需要使用高效的技术和算法，以提高请求的响应速度和处理能力。Go语言以其高效性和并发性而闻名，因此成为处理高并发请求的优秀选择。Go中有多种模式可供选择，如基于goroutine和channel的并发模型、使用池技术的协程模型等，以便根据具体应用的需要来选择适合的技术模式。

本文代码参考搬至

http://marcio.io/2015/07/handling-1-million-requests-per-minute-with-golang/

W1

W1 结构体类型，它有五个成员：

WgSend 用于等待任务发送的 goroutine 完成。
Wg 用于等待任务处理的 goroutine 完成。
MaxNum 表示 goroutine 池的大小。
Ch 是一个字符串类型的通道，用于传递任务。
DispatchStop 是一个空结构体类型的通道，用于停止任务分发。

type W1 struct {
	WgSend       *sync.WaitGroup
	Wg           *sync.WaitGroup
	MaxNum       int
	Ch           chan string
	DispatchStop chan struct{}
}

接下来是 Dispatch 方法，它将任务发送到通道 Ch 中。它通过 for 循环来发送 10 倍于 MaxNum 的任务，每个任务都是一个 goroutine。defer 语句用于在任务完成时减少 WgSend 的计数。select 语句用于在任务分发被中止时退出任务发送。

Dispatch

func (w *W1) Dispatch(job string) {
	w.WgSend.Add(10 * w.MaxNum)
	for i := 0; i < 10*w.MaxNum; i++ {
		go func(i int) {
			defer w.WgSend.Done()
			select {
			case w.Ch <- fmt.Sprintf("%d", i):
				return
			case <-w.DispatchStop:
				fmt.Println("退出发送 job: ", fmt.Sprintf("%d", i))
				return
			}
		}(i)
	}
}

StartPool

然后是 StartPool 方法，它创建了一个 goroutine 池来处理从通道 Ch 中读取到的任务。

如果通道 Ch 还没有被创建，那么它将被创建。如果计数器 WgSend 还没有被创建，那么它也将被创建。如果计数器 Wg 还没有被创建，那么它也将被创建。

如果通道 DispatchStop 还没有被创建，那么它也将被创建。

for 循环用于创建 MaxNum 个 goroutine 来处理从通道中读取到的任务。defer 语句用于在任务完成时减少 Wg 的计数。

func (w *W1) StartPool() {
	if w.Ch == nil {
		w.Ch = make(chan string, w.MaxNum)
	}
	if w.WgSend == nil {
		w.WgSend = &sync.WaitGroup{}
	}
	if w.Wg == nil {
		w.Wg = &sync.WaitGroup{}
	}
	if w.DispatchStop == nil {
		w.DispatchStop = make(chan struct{})
	}
	w.Wg.Add(w.MaxNum)
	for i := 0; i < w.MaxNum; i++ {
		go func() {
			defer w.Wg.Done()
			for v := range w.Ch {
				fmt.Printf("完成工作: %s \n", v)
			}
		}()
	}
}

Stop

最后是 Stop 方法，它停止任务分发并等待所有任务完成。

它关闭了通道 DispatchStop，等待 WgSend 中的任务发送 goroutine 完成，然后关闭通道 Ch，等待 Wg 中的任务处理 goroutine 完成。

func (w *W1) Stop() {
	close(w.DispatchStop)
	w.WgSend.Wait()
	close(w.Ch)
	w.Wg.Wait()
}

Go高级特性探究之处理1分钟百万请求详解

W2

SubWorker

type SubWorker struct {
	JobChan chan string
}

子协程，它有一个 JobChan，用于接收任务。

Run：SubWorker 的方法，用于启动一个子协程，从 JobChan 中读取任务并执行。

func (sw *SubWorker) Run(wg *sync.WaitGroup, poolCh chan chan string, quitCh chan struct{}) {
	if sw.JobChan == nil {
		sw.JobChan = make(chan string)
	}
	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			poolCh <- sw.JobChan
			select {
			case res := <-sw.JobChan:
				fmt.Printf("完成工作: %s \n", res)
			case <-quitCh:
				fmt.Printf("消费者结束...... \n")
				return
			}
		}
	}()
}

W2

type W2 struct {
	SubWorkers []SubWorker
	Wg         *sync.WaitGroup
	MaxNum     int
	ChPool     chan chan string
	QuitChan   chan struct{}
}

Dispatch

Dispatch：W2 的方法，用于从 ChPool 中获取 TaskChan，将任务发送给一个 SubWorker 执行。

func (w *W2) Dispatch(job string) {
	jobChan := <-w.ChPool
	select {
	case jobChan <- job:
		fmt.Printf("发送任务 : %s 完成 \n", job)
		return
	case <-w.QuitChan:
		fmt.Printf("发送者（%s）结束 \n", job)
		return
	}
}

StartPool

StartPool：W2 的方法，用于初始化协程池，启动所有子协程并把 TaskChan 存储在 ChPool 中。

func (w *W2) StartPool() {
	if w.ChPool == nil {
		w.ChPool = make(chan chan string, w.MaxNum)
	}
	if w.SubWorkers == nil {
		w.SubWorkers = make([]SubWorker, w.MaxNum)
	}
	if w.Wg == nil {
		w.Wg = &sync.WaitGroup{}
	}
	for i := 0; i < len(w.SubWorkers); i++ {
		w.SubWorkers[i].Run(w.Wg, w.ChPool, w.QuitChan)
	}
}

Stop

Stop：W2 的方法，用于停止协程的工作，并等待所有协程结束。

func (w *W2) Stop() {
	close(w.QuitChan)
	w.Wg.Wait()
	close(w.ChPool)
}

DealW2 函数则是整个协程池的入口，它通过 NewWorker 方法创建一个 W2 实例，然后调用 StartPool 启动协程池，并通过 Dispatch 发送任务，最后调用 Stop 停止协程池。

func DealW2(max int) {
	w := NewWorker(w2, max)
	w.StartPool()
	for i := 0; i < 10*max; i++ {
		go w.Dispatch(fmt.Sprintf("%d", i))
	}
	w.Stop()
}

Go高级特性探究之处理1分钟百万请求详解

个人见解

看到这里对于w2我已经有点迷糊了，还能传递w.Wg, w.ChPool, w.QuitChan?

原来是golang里如果方法传递的不是地址，那么就会做一个拷贝，所以这里调用的wg根本就不是一个对象。

传递的地方传递地址就可以了,如果不传递地址，将会出现死锁

go doSomething(i, &wg, ch)
func doSomething(index int, wg *sync.WaitGroup, ch chan int) {

w1也有一个比较大的问题。在处理请求时，每个 Goroutine 都会占用一定的系统资源，如果请求量过大，会造成 Goroutine 数量的剧增，消耗过多系统资源，程序可能会崩溃

能不能用上我之前写的协程池探究 Go 的高级特性之【Go 协程池】

探究原文

在这段代码中，poolCh代表工作者池，sw.JobChan代表工作者的工作通道。当一个工作者完成了工作后，它会将工作结果发送到sw.JobChan，此时可以通过case res := <-sw.JobChan:来接收该工作的结果。

在这个代码块中，还需要处理一个退出信号quitCh。因此，第二个case <-quitCh:用于检测是否接收到了退出信号。如果接收到了退出信号，程序将打印出消息并结束。

需要注意的是，这两个case语句是互斥的，只有当工作者完成工作或收到退出信号时，才会进入其中一个语句。因此，这个循环可以保证在工作者完成工作或收到退出信号时退出。

需要读取两次sw.JobChan的原因是：第一次读取用于将工作者的工作通道放回工作者池中，这样其他工作者就可以使用该通道。第二次读取用于接收工作者的工作结果或退出信号。因此，这两次读取是为了确保能够在正确的时刻将工作者的工作通道放回工作者池中并正确地处理工作结果或退出信号。

根据w2的特点我自己写了一个w2

import (
   "fmt"
   "sync"
)
type SubWorkerNew struct {
   JobChan chan string
}
type W2New struct {
   SubWorkers []SubWorkerNew
   Wg         *sync.WaitGroup
   MaxNum     int
   ChPool     chan chan string
   QuitChan   chan struct{}
}
func NewW2(maxNum int) *W2New {
   subWorkers := make([]SubWorkerNew, maxNum)
   for i := 0; i < maxNum; i++ {
      subWorkers[i] = SubWorkerNew{JobChan: make(chan string)}
   }
   pool := make(chan chan string, maxNum)
   for i := 0; i < maxNum; i++ {
      pool <- subWorkers[i].JobChan
   }
   return &W2New{
      SubWorkers: subWorkers,
      Wg:         &sync.WaitGroup{},
      MaxNum:     maxNum,
      ChPool:     pool,
      QuitChan:   make(chan struct{}),
   }
}
func (w *W2New) Dispatch(job string) {
   select {
   case jobChannel := <-w.ChPool:
      jobChannel <- job
   default:
      fmt.Println("All workers busy")
   }
}
func (w *W2New) StartPool() {
   for i := 0; i < w.MaxNum; i++ {
      go func(subWorker *SubWorkerNew) {
         w.Wg.Add(1)
         defer w.Wg.Done()
         for {
            select {
            case job := <-subWorker.JobChan:
               fmt.Println("processing ", job)
            case <-w.QuitChan:
               return
            }
         }
      }(&w.SubWorkers[i])
   }
}
func (w *W2New) Stop() {
   close(w.QuitChan)
   w.Wg.Wait()
   close(w.ChPool)
   for _, subWorker := range w.SubWorkers {
      close(subWorker.JobChan)
   }
}
func main() {
   w := NewW2(5)
   w.StartPool()
   for i := 0; i < 20; i++ {
      w.Dispatch(fmt.Sprintf("job %d", i))
   }
   w.Stop()
}

但是有几个点需要注意

1.没有考虑JobChan通道的缓冲区大小，如果有大量任务被并发分配，容易导致内存占用过高；

2.每个线程都会执行无限循环，此时线程退出的条件是接收到QuitChan通道的信号，可能导致线程的阻塞等问题；

3.Dispatch函数的默认情况下只会输出"All workers busy"，而不是阻塞，这意味着当所有线程都处于忙碌状态时，任务会丢失

4.线程池启动后无法动态扩展或缩小。

优化

这个优化版本改了很多次。有一些需要注意的点是,不然会一直死锁

1.使用sync.WaitGroup来确保线程池中所有线程都能够启动并运行；

2.在Stop函数中，先向SubWorker的JobChan中发送一个关闭信号，再等待所有SubWorker线程退出；

3.在Dispatch函数中，将默认情况下的输出改为阻塞等待可用通道；

w2new

package handle_million_requests
import (
	"fmt"
	"sync"
	"time"
)
type SubWorkerNew struct {
	Id      int
	JobChan chan string
}
type W2New struct {
	SubWorkers []SubWorkerNew
	MaxNum     int
	ChPool     chan chan string
	QuitChan   chan struct{}
	Wg         *sync.WaitGroup
}
func NewW2(maxNum int) *W2New {
	chPool := make(chan chan string, maxNum)
	subWorkers := make([]SubWorkerNew, maxNum)
	for i := 0; i < maxNum; i++ {
		subWorkers[i] = SubWorkerNew{Id: i, JobChan: make(chan string)}
		chPool <- subWorkers[i].JobChan
	}
	wg := new(sync.WaitGroup)
	wg.Add(maxNum)
	return &W2New{
		MaxNum:     maxNum,
		SubWorkers: subWorkers,
		ChPool:     chPool,
		QuitChan:   make(chan struct{}),
		Wg:         wg,
	}
}
func (w *W2New) StartPool() {
	for i := 0; i < w.MaxNum; i++ {
		go func(wg *sync.WaitGroup, subWorker *SubWorkerNew) {
			defer wg.Done()
			for {
				select {
				case job := <-subWorker.JobChan:
					fmt.Printf("SubWorker %d processing job %s\n", subWorker.Id, job)
					time.Sleep(time.Second) // 模拟任务处理过程
				case <-w.QuitChan:
					return
				}
			}
		}(w.Wg, &w.SubWorkers[i])
	}
}
func (w *W2New) Stop() {
	close(w.QuitChan)
	for i := 0; i < w.MaxNum; i++ {
		close(w.SubWorkers[i].JobChan)
	}
	w.Wg.Wait()
}
func (w *W2New) Dispatch(job string) {
	select {
	case jobChan := <-w.ChPool:
		jobChan <- job
	default:
		fmt.Println("All workers busy")
	}
}

func (w *W2New) AddWorker() {
	newWorker := SubWorkerNew{Id: w.MaxNum, JobChan: make(chan string)}
	w.SubWorkers = append(w.SubWorkers, newWorker)
	w.ChPool <- newWorker.JobChan
	w.MaxNum++
	w.Wg.Add(1)
	go func(subWorker *SubWorkerNew) {
		defer w.Wg.Done()
		for {
			select {
			case job := <-subWorker.JobChan:
				fmt.Printf("SubWorker %d processing job %s\n", subWorker.Id, job)
				time.Sleep(time.Second) // 模拟任务处理过程
			case <-w.QuitChan:
				return
			}
		}
	}(&newWorker)
}
func (w *W2New) RemoveWorker() {
	if w.MaxNum > 1 {
		worker := w.SubWorkers[w.MaxNum-1]
		close(worker.JobChan)
		w.MaxNum--
		w.SubWorkers = w.SubWorkers[:w.MaxNum]
	}
}

AddWorker和RemoveWorker，用于动态扩展/缩小线程池。

在AddWorker函数中，我们首先将MaxNum增加了1，然后创建一个新的SubWorkerNew结构体，将其添加到SubWorkers中，并将其JobChan通道添加到ChPool通道中。最后，我们创建一个新的协程来处理新添加的SubWorkerNew并让它进入无限循环，等待接收任务。
在RemoveWorker函数中，我们首先将MaxNum减少1，然后获取最后一个SubWorkerNew结构体，将它的JobChan通道发送到ChPool通道中，并从其通道中读取任何待处理的任务，最后创建一个新的协程来处理SubWorkerNew，继续处理任务。

测试用例

func TestW2New(t *testing.T) {  
    pool := NewW2(3)  
    pool.StartPool()  
    pool.Dispatch("task 1")  
    pool.Dispatch("task 2")  
    pool.Dispatch("task 3")  
    pool.AddWorker()  
    pool.AddWorker()  
    pool.RemoveWorker()  
    pool.Stop()  
}

Go高级特性探究之处理1分钟百万请求详解

当Dispatch函数向ChPool通道获取可用通道时，会从通道中取出一个SubWorker的JobChan通道，并将任务发送到该通道中。而对于SubWorker来说，并没有进行任务的使用次数限制，所以它可以处理多个任务。

在这个例子中，当任务数量比SubWorker数量多时，一个SubWorker的JobChan通道会接收到多个任务，它们会在SubWorker的循环中按顺序依次处理，直到JobChan中没有未处理的任务为止。因此，如果任务数量特别大，可能会导致某些SubWorker的JobChan通道暂时处于未处理任务状态，而其他的SubWorker在执行任务。

在测试结果中，最后三行中出现了多个"SubWorker 0 processing job"，说明SubWorker 0的JobChan通道接收了多个任务，并且在其循环中处理这些任务。下面的代码片段显示了这个过程：

SubWorker 0 的循环部分

for {
    select {
    case job := <-subWorker.JobChan:
        fmt.Printf("SubWorker %d processing job %s\n", subWorker.Id, job)
    case <-w.QuitChan:
        return
    }
}

Go高级特性探究之处理1分钟百万请求详解

声明：本站所有文章，如无特殊说明或标注，均为本站原创发布。任何个人或组织，在未征得本站同意时，禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益，可联系我们进行处理。

Go高级特性探究之处理1分钟百万请求详解

目录

W1

Dispatch

StartPool

Stop

W2

SubWorker

W2

Dispatch

StartPool

Stop

个人见解

探究原文

优化

w2new

测试用例

评论(0)

提示：请文明发言取消回复

作者信息

本站推荐

2025自媒体实战营，Deepseek、即梦 AI、Midjourney系统教程及案例实操

视频号短视频带货新玩法，用这个方法，一天佣金4407（附详细教程）

2025电商爆款课程，内销外销等方法，从0到爆单助力商家快速起量

2025小店运营全套系列课从小店基础入门到进阶精通

主播起号0-1快速成长从基础入门到进阶线上课

Ai五阶体系课入门通识理论+实操带你快速入门

热门资源

苹果cms海螺影视模板/大橙子模板/仿B站模板/v7模板/带手机移动端+详细安装使用说明

【已测】修复版H5骰子微信竞猜游戏骰宝免公众号版修复登录ID相同完美全套源码对接免签支付

网页游戏卧龙吟一键服务端加远程工具带架设教程

邪风曲单机版 2D回合制网络游戏源码一键安装即玩服务端公益服+GM工具

完整可用版本去水印小程序源码带教程源码

仙侠H5【苍穹剑诀】一键即玩端+授权后台+外网教程

Go高级特性探究之处理1分钟百万请求详解

目录

W1

Dispatch

StartPool

Stop

W2

SubWorker

W2

Dispatch

StartPool

Stop

个人见解

探究原文

优化

w2new

测试用例

评论(0)

提示：请文明发言 取消回复

相关文章

作者信息

本站推荐

热门资源

提示：请文明发言取消回复