Golang関数コルーチンプール実装技術共有-Golang-php.cn

Golang関数コルーチンプール実装技術共有

王林

リリース： 2023-05-16 08:31:52

オリジナル

2873 人が閲覧しました

Golang は、迅速な開発と同時実行性の高い言語として、当然ながらコルーチンプールの実装も備えています。コルーチンプールはコルーチンの管理に使用されるデータ構造であり、コルーチンの総数を制限し、コルーチンの作成と破棄のタイミングを制御することで、同時環境でのリソースの使用を最適化できます。次に、Golang の関数を使用してコルーチンプールを実装する方法を紹介します。

コルーチンプールの概念

コルーチンプールは、コルーチンの管理に使用されるデータ構造です。その目的は、コルーチンの数を制限し、コルーチンの作成と破棄のタイミングを調整し、プログラムの同時実行性を向上させます。

同時実行性が高い場合、コルーチンを開始すると毎回大量のオーバーヘッドが発生します。プログラムが同時に数百または数千のコルーチンを開く必要がある場合、これらのオーバーヘッドは非常に大きくなります。一般的な接続プールやスレッドプールと同様に、コルーチンプールはコンピューターリソースをより効率的に利用し、多数の同時操作を伴うタスクを完了できます。

コルーチンプールの実装アイデア

コルーチンプールは、スケーラブルプールと固定プールに分類できます。このうち、スケーラブルプールは需要に応じて自動的に容量を拡張・縮小できるのに対し、固定プールは最初から容量が固定されており変更できません。

コルーチンプールを実装する Golang 関数の主なアイデアは、2 つのチャネルを通じて通信することです。 1 つは、コルーチンワーカーにタスクを割り当てるために使用される workerChannel であり、もう 1 つは、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()
}

ログイン後にコピー

上記のコードを実行すると、次のことがわかります。コントロールステーションが出力するログ情報。このうち、ワーカー開始とは各ワーカーが実行を開始すること、新規ジョブとはタスクチャネルにタスクを追加すること、追加ジョブとはタスクがワーカーチャネルに配置されて実行を待機すること、実行後ジョブとはタスクが実行を開始することを意味します。正常に実行されました。

コード分析

上記のコードでは、NewPool 関数を使用して、タスクチャネル、ワーカーチャネル、ワーカーの数を含むコルーチンプールを初期化します。 Worker タイプはコルーチンワーカーに対応し、ワーカーコルーチンの実行を終了するタスクチャネルと終了チャネルが含まれます。 NewWorker 関数は、ワーカーオブジェクトを初期化し、そのタスクチャネルをコルーチンプール内のワーカーチャネルに追加する役割を果たします。

AddTask 関数は、コルーチンプールタスクチャネルに新しいタスクを追加するために使用されます。この関数は、タスクが追加されるまでブロックします。ワーカーチャネルに空きワーカーがある場合、タスクはそのワーカーに直接割り当てられます。そうでない場合は、ワーカーチャネル内のワーカーが解放されるまで待機します。

Start 関数は、ワーカーコルーチンを開始し、タスクの到着の待機を開始する役割を果たします。この関数は、まず独自のタスクチャネルをワーカーチャネルに追加し、次にタスクチャネルが閉じるかチャネル終了信号を受信するまでタスクの到着を待ちます。タスクを受信した場合は、タスクを実行します。ループ内でチャネル終了信号を受信した場合、コルーチンの実行を終了する必要があることを意味し、この時点でワーカーはワーカーチャネルから自身を削除します。

ディスパッチ関数は、タスクチャネルをリッスンし、利用可能なワーカーに基づいてタスクをチャネルに割り当てる go コルーチンです。タスクチャネルに新しいタスクがある場合、ディスパッチはワーカーチャネルから空きワーカーを取得し、それらにタスクを割り当てようとします。ワーカーチャネルに空きワーカーがない場合、ワーカーが解放されるまで待機します。