解析Go语言的高级特性

第第页解析Go语言的高级特性：Goland猫

对于大型的互联网应用程序，如电商平台、社交（网络）、金融交易平台等，每秒钟都会收到大量的请求。在这些应用程序中，需要使用高效的技术来应对高并发的请求，尤其是在短时间内处理大量的请求，如1分钟百万请求。

同时，为了降低用户的使用门槛和提升用户体验，前端需要实现参数的无感知传递。这样用户在使用时，无需担心参数传递的问题，能够轻松地享受应用程序的服务。

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

本文代码参考搬至

W1

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

WgSend用于等待任务发送的goroutine完成。

Wg用于等待任务处理的goroutine完成。

MaxNum表示goroutine池的大小。

Ch是一个字符串类型的通道，用于传递任务。

DispatchStop是一个空结构体类型的通道，用于停止任务分发。

type

W1

struct

{

WgSend

*sync.W（ai）tGroup

Wg

*sync.WaitGroup

MaxNum

int

Ch

chan

string

DispatchStop

chan

struct{}}

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

Dispatch

func

(w

*W1)

Dispatch(job

string)

{

w.WgSend.（Ad）d(10

*

w.MaxNum)

for

i

:=

0;

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

=

i

Stop

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

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

func

(w

*W1)

Stop()

{

close(w.DispatchStop)

w.WgSend.Wait()

close(w.Ch)

w.Wg.Wait()}

W2

（Sub）Worker

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

W2

type

W2

struct

{

SubWorke（rs）

[]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

:=

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

=

i

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

个人见解

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

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

doSomething(i,

i

但是有几个点需要注意

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

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

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

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

优化

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

1.使用sync.WaitGroup来确保线程池中所有线程都能够启动并运行；2.在Stop函数中，先向SubWorker的JobChan中发送一个关闭信号，再等待所有SubWorker线程退出；3.在Dispatch函数中，将默认情况下的输出改为阻塞等待可用通道；

w2new

package

handle_million_requestsimport

(

"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

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

（Te）stW2New(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()

}

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

在这个例子中，当任务数量比SubWorker数量多时，一个SubWorker的JobChan通道会接收到多个任务，它们会在SubWorker的循环中按顺序依次处理，直到JobChan中没有未处理的任务为止。因此，如