火了，忘了戈罗丁戈朗

Question

我编写了一个 API，用于进行数据库调用并执行一些业务逻辑。我正在调用一个必须在后台执行某些操作的戈鲁廷。由于 API 调用不应等待此后台任务完成，因此我在调用 goroutine 后立即返回 200 OK（让我们假设后台任务永远不会给出任何错误）。我读到戈鲁廷将在戈鲁廷完成其任务后被终止。这种火灾和忘记远离戈鲁丁泄漏是安全的吗？一旦戈鲁廷执行了工作，它们是否会被终止和清理？func DefaultHandler(w http.ResponseWriter, r *http.Request) {    // Some DB calls    // Some business logics    go func() {        // some Task taking 5 sec    }()    w.WriteHeader(http.StatusOK)}

Answer 1

我建议始终控制您的戈鲁丁，以避免内存和系统耗尽。如果您收到请求激增，并且开始不受控制地生成goroutines，则系统可能很快就会关闭。

在那些需要立即返回200Ok的情况下，最好的方法是创建一个消息队列，因此服务器只需要在队列中创建一个作业并返回ok和忘记。其余部分将由使用者异步处理。

创建者（HTTP 服务器）>>>队列>>>使用者

通常，队列是外部资源 （RabbitMQ， AWS SQS...），但出于教学目的，您可以使用与消息队列相同的通道实现相同的效果。

在示例中，您将看到我们如何创建一个通道来传达 2 个进程。然后，我们启动将从通道读取的工作进程，然后使用将写入通道的处理程序启动服务器。

尝试在发送 curl 请求时使用缓冲区大小和作业时间。

package main

import (

    "fmt"

    "log"

    "net/http"

    "time"

)

/*

$ go run .

curl "http://localhost:8080?user_id=1"

curl "http://localhost:8080?user_id=2"

curl "http://localhost:8080?user_id=3"

curl "http://localhost:8080?user_id=....."

*/

func main() {

    queueSize := 10

    // This is our queue, a channel to communicate processes. Queue size is the number of items that can be stored in the channel

    myJobQueue := make(chan string, queueSize) // Search for 'buffered channels'

    // Starts a worker that will read continuously from our queue

    go myBackgroundWorker(myJobQueue)

    // We start our server with a handler that is receiving the queue to write to it

    if err := http.ListenAndServe("localhost:8080", myAsyncHandler(myJobQueue)); err != nil {

        panic(err)

    }

}

func myAsyncHandler(myJobQueue chan<- string) http.HandlerFunc {

    return func(rw http.ResponseWriter, r *http.Request) {

        // We check that in the query string we have a 'user_id' query param

        if userID := r.URL.Query().Get("user_id"); userID != "" {

            select {

            case myJobQueue <- userID: // We try to put the item into the queue ...

                rw.WriteHeader(http.StatusOK)

                rw.Write([]byte(fmt.Sprintf("queuing user process: %s", userID)))

            default: // If we cannot write to the queue it's because is full!

                rw.WriteHeader(http.StatusInternalServerError)

                rw.Write([]byte(`our internal queue is full, try it later`))

            }

            return

        }

        rw.WriteHeader(http.StatusBadRequest)

        rw.Write([]byte(`missing 'user_id' in query params`))

    }

}

func myBackgroundWorker(myJobQueue <-chan string) {

    const (

        jobDuration = 10 * time.Second // simulation of a heavy background process

    )

    // We continuosly read from our queue and process the queue 1 by 1.

    // In this loop we could spawn more goroutines in a controlled way to paralelize work and increase the read throughput, but i don't want to overcomplicate the example.

    for userID := range myJobQueue {

        // rate limiter here ...

        // go func(u string){

        log.Printf("processing user: %s, started", userID)

        time.Sleep(jobDuration)

        log.Printf("processing user: %s, finisehd", userID)

        // }(userID)

    }

}

Answer 2

您无需处理“戈鲁丁清理”，您只需启动戈鲁丁，当作为戈鲁廷返回的功能启动时，它们将被清除。引用规范：围棋声明：

当函数终止时，其戈鲁廷也终止。如果函数具有任何返回值，则在函数完成时将丢弃这些返回值。

所以你所做的很好。但请注意，您启动的 goroutine 不能使用或假定有关请求 （） 和响应编写器 （） 的任何内容，您只能在从处理程序返回之前使用它们。rw

另请注意，您不必编写 ，如果您从处理程序返回而不写入任何内容，则假定为成功并将自动发送回去。http.StatusOKHTTP 200 OK

查看相关/可能的重复项：在另一个戈鲁廷上运行的 Webhook 进程

Answer 3

没有“戈鲁丁清洁”，你可以使用网络钩子或像gocraft这样的后台工作。我能想到的使用你的解决方案的唯一方法是将同步包用于学习目的。

func DefaultHandler(w http.ResponseWriter, r *http.Request) {

// Some DB calls

// Some business logics

var wg sync.WaitGroup

wg.Add(1)

go func() {

  defer wg.Done()

    // some Task taking 5 sec

}()

w.WriteHeader(http.StatusOK)

wg.wait()

}

Answer 4

您可以等待戈鲁廷完成使用：&sync.WaitGroup

// BusyTask

func BusyTask(t interface{}) error {

    var wg = &sync.WaitGroup{}

    wg.Add(1)

    go func() {

        // busy doing stuff

        time.Sleep(5 * time.Second)

        wg.Done()

    }()

    wg.Wait() // wait for goroutine

    return nil

}

// this will wait 5 second till goroutune finish

func main() {

    fmt.Println("hello")

    BusyTask("some task...")

    fmt.Println("done")

}

另一种方法是将 a 附加到戈鲁丁并超时。context.Context

//

func BusyTaskContext(ctx context.Context, t string) error {

    done := make(chan struct{}, 1)

    //

    go func() {

        // time sleep 5 second

        time.Sleep(5 * time.Second)

        // do tasks and signle done

        done <- struct{}{}

        close(done)

    }()

    //

    select {

    case <-ctx.Done():

        return errors.New("timeout")

    case <-done:

        return nil

    }

}

//

func main() {

    fmt.Println("hello")

    ctx, cancel := context.WithTimeout(context.TODO(), 2*time.Second)

    defer cancel()

    if err := BusyTaskContext(ctx, "some task..."); err != nil {

        fmt.Println(err)

        return

    }

    fmt.Println("done")

}