Gemeinsame Nutzung der Golang-Funktions-Coroutine-Pool-Implementierungstechnologie

Golang verfügt als Sprache mit schneller Entwicklung und hoher Parallelität natürlich auch über die Implementierung eines Coroutine-Pools. Der Coroutine-Pool ist eine Datenstruktur zur Verwaltung von Coroutinen. Er kann die Gesamtzahl der Coroutinen begrenzen und den Zeitpunkt ihrer Erstellung und Zerstörung steuern, wodurch die Ressourcennutzung in einer gleichzeitigen Umgebung optimiert wird. Als Nächstes werde ich vorstellen, wie man Golang-Funktionen zum Implementieren von Coroutine-Pools verwendet.

1. Das Konzept des Coroutine-Pools

Der Coroutine-Pool ist eine Datenstruktur zur Verwaltung von Coroutinen. Der Zweck besteht darin, die Anzahl der Coroutinen zu begrenzen und den Zeitpunkt ihrer Erstellung und Zerstörung zu steuern, wodurch die Parallelität des Programms verbessert wird.

Bei hoher Parallelität erzeugt das Starten einer Coroutine jedes Mal viel Overhead. Wenn das Programm Hunderte oder Tausende von Coroutinen gleichzeitig öffnen muss, wird dieser Overhead sehr groß. Ähnlich wie herkömmliche Verbindungspools und Thread-Pools können Coroutine-Pools Computerressourcen besser nutzen und Aufgaben mit einer großen Anzahl gleichzeitiger Vorgänge erledigen.

2. Implementierungsideen des Coroutine-Pools

Der Coroutine-Pool kann in skalierbare Pools und feste Pools unterteilt werden. Unter anderem kann der skalierbare Pool die Kapazität entsprechend der Nachfrage automatisch erweitern und verkleinern, während der feste Pool zu Beginn eine feste Kapazität hat und nicht geändert werden kann.

Die Hauptidee der Golang-Funktion zur Implementierung des Coroutine-Pools besteht darin, über zwei Kanäle zu kommunizieren. Einer ist der workerChannel, der zum Zuweisen von Aufgaben an Coroutine-Worker verwendet wird, und der andere ist der Task-Kanal, der zum Übergeben von Aufgaben an den workerChannel verwendet wird. Wenn eine Aufgabe ausgeführt werden muss, wird die Aufgabe aus dem Aufgabenkanal entfernt und eine Coroutine basierend auf der Anzahl der verfügbaren Worker im WorkerChannel erstellt oder die Aufgabe wird direkt einem inaktiven Worker zur Ausführung zugewiesen. Arbeiter, die die Aufgabe erledigen, kehren zum workerChannel zurück und warten auf die nächste Aufgabenzuweisung. Natürlich kann der Coroutine-Pool in manchen Fällen auch weitere Datenstrukturen wie Mutexe oder Wartegruppen enthalten, um zu steuern, wie Aufgaben ausgeführt werden.

3. Implementierungscode des Coroutine-Pools

Das Folgende ist der spezifische Code zum Implementieren des Coroutine-Pools:

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

Durch Ausführen des obigen Codes können Sie die von der Konsole ausgegebenen Protokollinformationen sehen. Unter diesen bedeutet „Worker-Start“, dass jeder Worker mit der Ausführung beginnt, „Neuer Job“ bedeutet, dass eine Aufgabe zum Task-Kanal hinzugefügt wird, „Job anhängen“ bedeutet, dass die Aufgabe im Worker-Kanal platziert wird und auf die Ausführung wartet, und „Nach dem Ausführen“-Job bedeutet, dass die Aufgabe ausgeführt wurde erfolgreich ausgeführt.

4. Code-Analyse

Im obigen Code wird die NewPool-Funktion verwendet, um den Coroutine-Pool zu initialisieren, der Aufgabenkanäle, Worker-Kanäle und die Anzahl der Worker umfasst. Der Worker-Typ entspricht dem Coroutine-Worker und umfasst einen Task-Kanal und einen Quit-Kanal, um die Ausführung der Worker-Coroutine zu beenden. Die NewWorker-Funktion ist für die Initialisierung des Worker-Objekts und das Hinzufügen seines Aufgabenkanals zum Worker-Kanal im Coroutine-Pool verantwortlich.

Die AddTask-Funktion wird verwendet, um eine neue Aufgabe zum Coroutine-Pool-Aufgabenkanal hinzuzufügen. Diese Funktion blockiert, bis die Aufgabe hinzugefügt wird. Wenn im Worker-Kanal ein freier Worker vorhanden ist, wird die Aufgabe direkt dem Worker zugewiesen, andernfalls wird auf die Freigabe eines Workers im Worker-Kanal gewartet.

Die Startfunktion ist dafür verantwortlich, die Worker-Coroutine zu starten und auf das Eintreffen der Aufgabe zu warten. Diese Funktion fügt dem Worker-Kanal zunächst einen eigenen Aufgabenkanal hinzu und wartet dann auf das Eintreffen der Aufgabe, bis der Aufgabenkanal geschlossen wird oder das Signal zum Beenden des Kanals empfangen wird. Wenn eine Aufgabe empfangen wird, führen Sie die Aufgabe aus. Wenn das Signal zum Beenden des Kanals innerhalb der Schleife empfangen wird, bedeutet dies, dass die Ausführung der Coroutine beendet werden muss und der Worker sich zu diesem Zeitpunkt aus dem Worker-Kanal entfernt.

Die Dispatch-Funktion ist eine Go-Coroutine, die auf Aufgabenkanäle hört und ihnen basierend auf den verfügbaren Arbeitskräften Aufgaben zuweist. Wenn im Aufgabenkanal neue Aufgaben vorhanden sind, versucht der Disponent, freie Arbeitskräfte aus dem Arbeitskanal zu erhalten und ihnen die Aufgaben zuzuweisen. Wenn im Worker-Kanal keine freien Worker vorhanden sind, wird gewartet, bis ein Worker freigegeben wird.

5. Zusammenfassung

Dieser Artikel stellt die Idee und den Implementierungscode der Golang-Funktion zur Implementierung des Coroutine-Pools vor. Die Anzahl der Coroutinen kann über den Coroutine-Pool gesteuert werden, wodurch die Computerressourcen vollständig genutzt und die Programmparallelität in einer Umgebung mit hoher Parallelität verbessert werden.

Das obige ist der detaillierte Inhalt vonGemeinsame Nutzung der Golang-Funktions-Coroutine-Pool-Implementierungstechnologie. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!