Golang函数的协程池实现技术分享

王林原创: 2023-05-16 08:31:52 1460浏览

Golang作为一门快速开发、高并发的语言，自然也有协程池的实现方式。协程池是用于管理协程的数据结构，通过它可以限制协程的总数量、并控制其创建与销毁的时机，从而优化并发环境下资源的使用。接下来，我将介绍如何使用Golang函数实现协程池。

协程池的概念

协程池是一种用于管理协程的数据结构，目的是限制协程的数量并控制其创建与销毁的时机，从而提高程序的并发性。

在高并发的情况下，每次启动一个协程都会产生大量的开销。如果程序需要同时开启成百上千个协程，这些开销将会变得非常显著。类似于常见的连接池和线程池，协程池可以更好地利用计算机的资源，完成涉及大量并发操作的任务。

协程池的实现思路

协程池可以分为可扩展池和固定池。其中，可扩展池可根据需求自动扩展和缩小容量，固定池则是一开始固定容量，不可更改。

Golang函数实现协程池的主要思路是通过两个channel进行通信。一个是workerChannel，用于将任务分配给协程工人，另一个是任务channel，用于将任务传递给workerChannel。当有任务需要执行时，从任务channel中取出任务，并根据workerChannel中的可用工人数量进行协程的创建或直接将任务分配给空闲的工人进行执行。完成任务的工人会重新回到workerChannel，等待下一次任务的分配。当然，在一些情况下，协程池还可以包含更多的数据结构，例如互斥锁或者等待组，来控制任务的执行方式。

协程池的实现代码

下面是实现协程池的具体代码：

package main

import (
    "fmt"
    "sync"
)

type Task struct {
    f func() error
}

var wg sync.WaitGroup

type Pool struct {
    //任务通道
    JobQueue chan Task
    //worker通道
    WorkerQueue chan chan Task
    //worker数量
    MaxWorkers int
}

func NewPool(maxWorkers int) *Pool {
    return &Pool{
        JobQueue:    make(chan Task, 10),
        WorkerQueue: make(chan chan Task, maxWorkers),
        MaxWorkers:  maxWorkers,
    }
}

func (p *Pool) Run() {
    for i := 0; i < p.MaxWorkers; i++ {
        worker := NewWorker(i+1, p.WorkerQueue)
        worker.Start()
    }

    go p.dispatch()
}

func (p *Pool) dispatch() {
    for {
        select {
        case job := <-p.JobQueue:
            fmt.Println("new job")
            worker := <-p.WorkerQueue
            fmt.Println("append job")
            worker <- job
            fmt.Println("after run job")
        }
    }
}

func (p *Pool) AddTask(task Task) {
    p.JobQueue <- task
}

type Worker struct {
    id          int
    WorkerQueue chan chan Task
    JobChannel  chan Task
    quitChan    chan struct{}
}

func NewWorker(id int, workerQueue chan chan Task) Worker {
    fmt.Println("newWorker")
    return Worker{
        id:          id,
        WorkerQueue: workerQueue,
        JobChannel:  make(chan Task),
        quitChan:    make(chan struct{}),
    }
}

func (w *Worker) Start() {
    fmt.Println("worker start")
    go func() {
        for {
            //将自己的jobChannel放入worker队列中
            w.WorkerQueue <- w.JobChannel
            select {
            case task := <-w.JobChannel:
                fmt.Printf("worker%d start job
", w.id)
                task.f()
                fmt.Printf("worker%d finished job
", w.id)
            case <-w.quitChan:
                fmt.Printf("worker%d quit
", w.id)
                return
            }
        }
    }()
}

func (w *Worker) Stop() {
    go func() {
        w.quitChan <- struct{}{}
    }()
}

func Hello() error {
    fmt.Println("Hello World")
    wg.Done()
    return nil
}

func main() {
    p := NewPool(5)
    p.Run()

    for i := 0; i < 100; i++ {
        task := Task{
            f: Hello,
        }
        wg.Add(1)
        p.AddTask(task)
    }
    wg.Wait()
}

通过运行以上代码，可以看到控制台输出的日志信息。其中，worker start表示每个worker开始运行，new job代表新增任务到任务通道中，append job表示任务被放到worker通道中等待执行，after run job表示任务已被成功执行。

代码解析

上述代码中，NewPool函数用于初始化协程池，里面包含了任务通道、worker通道和worker数量。Worker类型对应协程工人，包含一个任务通道和一个quit channel用于结束worker协程的运行。NewWorker函数负责初始化worker对象，并将其任务通道加入协程池中的worker通道中。

AddTask函数用于向协程池任务通道中添加一个新任务。这个函数是阻塞的，直到任务被添加为止。如果worker通道中有空余的worker，则直接将任务分配给worker，否则将等待worker通道中的某个worker被释放。

Start函数负责启动worker协程并开始等待任务的到来。该函数会首先将自己的任务通道加入worker通道中，然后等待任务到来，直到任务通道被关闭或者收到quit channel的信号。如果收到任务，则执行任务。如果循环内收到了quit channel的信号，则表示需要结束该协程的运行，此时worker会将自己从worker通道中移除。

dispatch函数是一个go协程，用于监听任务通道，并根据空余的worker分配任务给它们。当任务通道中有新任务时，dispatch会尝试从worker通道中获取空余worker并将任务分配给它们。如果worker通道中没有空余的worker，则会一直等待，直到有worker被释放。