Partage de technologie de mise en œuvre du pool de coroutines de fonction Golang

Golang, en tant que langage à développement rapide et à haute concurrence, a naturellement également la mise en œuvre d'un pool de coroutines. Le pool de coroutines est une structure de données utilisée pour gérer les coroutines. Il peut limiter le nombre total de coroutines et contrôler le moment de leur création et de leur destruction, optimisant ainsi l'utilisation des ressources dans un environnement simultané. Ensuite, je présenterai comment utiliser les fonctions Golang pour implémenter des pools de coroutines.

1. Le concept de pool de coroutines

Le pool de coroutines est une structure de données utilisée pour gérer les coroutines. Le but est de limiter le nombre de coroutines et de contrôler le timing de leur création et de leur destruction, améliorant ainsi la simultanéité du programme.

Dans le cas d'une concurrence élevée, démarrer une coroutine générera à chaque fois beaucoup de surcharge. Si le programme doit ouvrir des centaines ou des milliers de coroutines en même temps, ces frais généraux deviendront très importants. Semblables aux pools de connexions et aux pools de threads courants, les pools de coroutines peuvent mieux utiliser les ressources informatiques et effectuer des tâches impliquant un grand nombre d'opérations simultanées.

2. Idées de mise en œuvre du pool coroutine

Le pool coroutine peut être divisé en pool évolutif et pool fixe. Parmi eux, le pool évolutif peut automatiquement augmenter et réduire la capacité en fonction de la demande, tandis que le pool fixe a une capacité fixe au début et ne peut pas être modifié.

L'idée principale de la fonction Golang pour implémenter le pool de coroutines est de communiquer via deux canaux. L'un est le travailleurChannel, qui est utilisé pour attribuer des tâches aux travailleurs de la coroutine, et l'autre est le canal de tâches, qui est utilisé pour transmettre des tâches au travailleurChannel. Lorsqu'une tâche doit être exécutée, la tâche est supprimée du canal de tâches et une coroutine est créée en fonction du nombre de travailleurs disponibles dans le canal de travail ou la tâche est directement affectée à un travailleur inactif pour exécution. Les travailleurs qui terminent la tâche reviendront au workerChannel et attendront la prochaine affectation de tâche. Bien entendu, dans certains cas, le pool de coroutines peut également contenir davantage de structures de données, telles que des mutex ou des groupes d'attente, pour contrôler la manière dont les tâches sont exécutées.

3. Code d'implémentation du pool de coroutines

Voici le code spécifique pour implémenter le pool de coroutines :

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()
}

En exécutant le code ci-dessus, vous pouvez voir les informations de journal générées par la console. Parmi eux, le démarrage du travailleur signifie que chaque travailleur commence à s'exécuter, le nouveau travail signifie l'ajout d'une tâche au canal des tâches, l'ajout du travail signifie que la tâche est placée dans le canal du travail et en attente d'exécution, et après l'exécution du travail signifie que la tâche a été exécuté avec succès.

4. Analyse du code

Dans le code ci-dessus, la fonction NewPool est utilisée pour initialiser le pool de coroutines, qui comprend les canaux de tâches, les canaux de travailleurs et le nombre de travailleurs. Le type Worker correspond au travailleur de la coroutine et comprend un canal de tâches et un canal de sortie pour terminer l'exécution de la coroutine de travail. La fonction NewWorker est chargée d'initialiser l'objet Worker et d'ajouter son canal de tâches au canal Worker dans le pool de coroutines.

La fonction AddTask est utilisée pour ajouter une nouvelle tâche au canal de tâches du pool de coroutines. Cette fonction se bloque jusqu'à ce que la tâche soit ajoutée. S'il y a un travailleur libre dans le canal des travailleurs, la tâche sera assignée directement au travailleur, sinon elle attendra qu'un travailleur dans le canal des travailleurs soit libéré.

La fonction Start est chargée de démarrer la coroutine du travailleur et de commencer à attendre l'arrivée de la tâche. Cette fonction ajoutera d'abord son propre canal de tâches au canal de travail, puis attendra que la tâche arrive jusqu'à ce que le canal de tâches soit fermé ou que le signal de sortie du canal soit reçu. Si une tâche est reçue, exécutez-la. Si le signal de sortie du canal est reçu dans la boucle, cela signifie que l'exécution de la coroutine doit se terminer et le travailleur se retirera du canal travailleur à ce moment-là.

La fonction de répartition est une coroutine go qui écoute les canaux de tâches et leur attribue des tâches en fonction des travailleurs disponibles. Lorsqu'il y a de nouvelles tâches dans le canal de tâches, la répartition essaiera d'obtenir des travailleurs libres du canal de travail et de leur attribuer les tâches. S'il n'y a aucun travailleur libre dans le canal des travailleurs, il attendra qu'un travailleur soit libéré.

5. Résumé

Cet article présente l'idée et le code d'implémentation de la fonction Golang pour implémenter le pool de coroutines. Le nombre de coroutines peut être contrôlé via le pool de coroutines, ce qui permet d'utiliser pleinement les ressources informatiques et d'améliorer la simultanéité des programmes dans un environnement à forte simultanéité.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!