So brechen Sie einen Kanaljob basierend auf der ID in Golang ab

In Golang ist die Kündigung von Channel-Jobs eine häufige Anforderung. Wenn wir mit gleichzeitigen Aufgaben arbeiten, müssen wir manchmal die ausgeführte Aufgabe basierend auf der Aufgaben-ID abbrechen. Wie implementiert man diese Funktion in Golang? Im Folgenden stelle ich Ihnen eine einfache und effektive Methode vor. Zuerst müssen wir einen gepufferten Kanal erstellen, um die ID der Aufgabe zu speichern. Anschließend können wir mit einer Select-Anweisung den Lesevorgang des Kanals abhören und die Task-ID an den Kanal senden, wenn die Task abgebrochen werden muss. Anschließend können wir in der Aufgabenausführungsfunktion feststellen, ob die Aufgabe abgebrochen werden muss, indem wir beurteilen, ob der Kanal geschlossen ist. Wenn der Kanal geschlossen ist, wurde die Aufgabe abgebrochen und wir können die Ausführung der Aufgabe an der entsprechenden Stelle beenden. Auf diese Weise können wir Channel-Jobs basierend auf der ID in Golang einfach stornieren. Das Obige ist die vom PHP-Editor Xinyi eingeführte Methode. Ich hoffe, sie wird für alle hilfreich sein!

Frageninhalt

Also ich habe eins post 端点，它创建一个作业并将它们添加到一个 chan 中。 workerjobschan = make(chan job, maxqueuesize)

So mache ich meinen Job im Kanal (main.go):

for i := 1; i <= maxworkers; i++ {
    go func(i int) {
        for job := range workerjobschan {
            ctx, cancel := context.withcancel(context.background())
            storejob(job.search.id, cancel)
            job.execute(ctx, c.db, i)
        }
    }(i)
}

Die Stornofunktion hinterlege ich in der Karte: canceljobfuncs = make(map[int]context.cancelfunc).

Das ist die Jobfunktion:

func (j *job) execute(ctx context.context, db *sql.db, workerid int) error {


    for {
        select {
        // check for cancellation signal
        case <-ctx.done():
             if err := ctx.err(); err != nil {
                fmt.println("worker", workerid, "error", err)
             }
            fmt.println("worker", workerid, "cancelled")
            return nil

        default:
            fmt.printf("worker%d: processing %s\n", workerid, j.search.query)
            time.sleep(2 * time.second)
            fmt.printf("worker%d: active %s\n", workerid, j.search.query)
            time.sleep(5 * time.second)
            fmt.printf("worker%d: completed %s!\n", workerid, j.search.query)

        }
    }
}

Ich breche den Kontext (im http-Handler) wie folgt ab:

cancelJob(search.ID)

Aber der Job läuft weiter. Ich habe viele Dinge ausprobiert, aber es scheint, als ob es nicht funktioniert.

Problemumgehung

Hier ist eine Möglichkeit, einen offensichtlichen Punkt zu veranschaulichen: Wenn Ihr Code nicht nach ctx.done() sucht, kann er nicht erkennen, dass er abgebrochen wurde.
(Übrigens ist dies eine weitere Umschreibung dessen, was @jimb im Kommentar zu Ihrer Frage geschrieben hat).

Wenn also der Code in der .execute(...)-Methode beginnt, diesen Block auszuführen:

fmt.printf("worker%d: processing %s\n", workerid, j.search.query)
    time.sleep(2 * time.second)
    fmt.printf("worker%d: active %s\n", workerid, j.search.query)
    time.sleep(5 * time.second)
    fmt.printf("worker%d: completed %s!\n", workerid, j.search.query)

Das Ende des Blocks wird erreicht (7 Sekunden). Es gibt keine Anweisung, beim Abbrechen anzuhalten.

Wenn Sie möchten, dass Ihre Funktion einen Abbruch während einer „Ruhe“-Anweisung erkennt, müssen Sie Ihren Code ändern.

Hier ist ein Beispiel dafür anhand Ihres Beispiels:

func (j *job) execute(ctx context.context, db *sql.db, workerid int) error {


    for {
        fmt.printf("worker%d: processing %s\n", workerid, j.search.query)

        // rewrite time.sleep() with time.after() so that it can be composed
        // in a select statement:
        select {
        case <-ctx.done():
            fmt.println("worker", workerid, "cancelled")
            return nil
        case <-time.after(2 * time.second):
            // keep going
        }

        fmt.printf("worker%d: active %s\n", workerid, j.search.query)

        select {
        case <-ctx.done():
            fmt.println("worker", workerid, "cancelled")
            return nil
        case <-time.after(5 * time.second):
            // keep going
        }

        fmt.printf("worker%d: completed %s!\n", workerid, j.search.query)
    }
}

https://www.php.cn/link/3bc31a430954d8326605fc690ed22f4d

Ich vermute, Ihr tatsächlicher Code hat kein time.sleep() 指令，而是 processsearch(...) 或 doquery(...) oder ...

Wenn Sie möchten, dass diese Funktionen während der Ausführung abgebrochen werden können, müssen Sie den Abbruchkontext irgendwie an sie übergeben und sie auf irgendeine Weise auf Abbruch prüfen lassen.

Eine Möglichkeit, den Kontext zu „übergeben“, besteht offensichtlich darin, ihn zu den Parametern der Funktion hinzuzufügen:

processsearch(ctx, ...)
doquery(ctx, ...)

Aber abhängig von Ihrem vorhandenen Code verfügen einige Parameter möglicherweise bereits über integrierte Methoden zum Abbrechen.
Zum Beispiel:

// an http.Request carries a context:
func doQuery(req *http.Request, ....) {
    ...
}

// at call site:
    ...
    req := http.NewRequestWithContext(ctx, "GET", "https://some.other.service/", nil)
    doQuery(req, ...)

Das obige ist der detaillierte Inhalt vonSo brechen Sie einen Kanaljob basierend auf der ID in Golang ab. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!