Golang 함수 코루틴 풀 구현 기술 공유-Golang-php.cn

Golang 함수 코루틴 풀 구현 기술 공유

王林

풀어 주다： 2023-05-16 08:31:52

원래의

2909명이 탐색했습니다.

Golang은 빠르게 개발되고 동시성이 높은 언어로서 당연히 코루틴 풀도 구현합니다. 코루틴 풀은 코루틴을 관리하는 데 사용되는 데이터 구조입니다. 코루틴의 총 개수를 제한하고 생성 및 소멸 시점을 제어하여 동시 환경에서 리소스 사용을 최적화할 수 있습니다. 다음으로 Golang 함수를 사용하여 코루틴 풀을 구현하는 방법을 소개하겠습니다.

코루틴 풀의 개념

코루틴 풀은 코루틴을 관리하는 데 사용되는 데이터 구조입니다. 목적은 코루틴 수를 제한하고 생성 및 소멸 시기를 제어하여 프로그램의 동시성을 향상시키는 것입니다.

동시성이 높은 경우 코루틴을 시작하면 매번 많은 오버헤드가 발생합니다. 프로그램이 동시에 수백 또는 수천 개의 코루틴을 열어야 하는 경우 이러한 오버헤드는 매우 커집니다. 일반적인 연결 풀 및 스레드 풀과 마찬가지로 코루틴 풀은 컴퓨터 리소스를 더 잘 활용하고 많은 수의 동시 작업이 포함된 작업을 완료할 수 있습니다.

코루틴 풀 구현 아이디어

코루틴 풀은 확장 가능한 풀과 고정 풀로 나눌 수 있습니다. 그 중 확장 가능한 풀은 수요에 따라 용량을 자동으로 확장 및 축소할 수 있는 반면, 고정 풀은 처음에 고정된 용량을 가지며 변경할 수 없습니다.

코루틴 풀을 구현하는 Golang 함수의 주요 아이디어는 두 가지 채널을 통해 통신하는 것입니다. 하나는 코루틴 워커에 작업을 할당하는 데 사용되는 WorkerChannel이고, 다른 하나는 WorkerChannel에 작업을 전달하는 데 사용되는 작업 채널입니다. 실행해야 할 작업이 있으면 작업 채널에서 해당 작업을 꺼내어 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)은 각 작업자가 실행을 시작한다는 의미이고, 새 작업(new job)은 작업 채널에 작업을 추가하는 것을 의미하고, 추가 작업(append job)은 작업이 작업자 채널에 배치되어 실행을 기다리는 것을 의미하며, 실행 후 작업(After run job)은 작업이 완료되었음을 의미합니다. 성공적으로 실행되었습니다.

코드 분석

위 코드에서 NewPool 함수는 작업 채널, 작업자 채널 및 작업자 수가 포함된 코루틴 풀을 초기화하는 데 사용됩니다. Worker 유형은 코루틴 작업자에 해당하며 작업 채널과 작업자 코루틴 실행을 종료하는 종료 채널을 포함합니다. NewWorker 함수는 작업자 개체를 초기화하고 해당 작업 채널을 코루틴 풀의 작업자 채널에 추가하는 일을 담당합니다.

AddTask 함수는 코루틴 풀 작업 채널에 새 작업을 추가하는 데 사용됩니다. 이 함수는 작업이 추가될 때까지 차단됩니다. 작업자 채널에 무료 작업자가 있으면 작업이 작업자에게 직접 할당되고, 그렇지 않으면 작업자 채널의 작업자가 해제될 때까지 기다립니다.

Start 함수는 작업자 코루틴을 시작하고 작업 도착을 기다리기 시작하는 역할을 담당합니다. 이 함수는 먼저 자체 작업 채널을 작업자 채널에 추가한 다음 작업 채널이 닫히거나 채널 종료 신호가 수신될 때까지 작업이 도착할 때까지 기다립니다. 작업이 수신되면 작업을 실행합니다. 루프 내에서 채널 종료 신호가 수신되면 코루틴 실행을 종료해야 하며 이때 작업자는 작업자 채널에서 스스로 제거된다는 의미입니다.

디스패치 기능은 작업 채널을 수신하고 사용 가능한 작업자를 기반으로 작업을 할당하는 go 코루틴입니다. 작업 채널에 새 작업이 있으면 파견은 작업자 채널에서 작업자를 확보하고 작업을 할당하려고 시도합니다. 작업자 채널에 사용 가능한 작업자가 없으면 작업자가 해제될 때까지 기다립니다.