Golang HTTP请求工作池

问题描述 投票：-1回答：3

我正在尝试构建一个系统，工作池/作业队列，以便在每个API端点上处理尽可能多的http requests。我调查了这个example并让它工作得很好，除了我偶然发现了我不明白如何将pool / jobqueue扩展到不同端点的问题。

为了方案起见，让我们描绘一个Golang http服务器，它在不同的端点和请求类型GET和POST ETC上有一百万请求/分钟。

我该如何扩展这个概念？我应该为每个端点创建不同的工作池和作业。或者我可以创建不同的作业并将它们输入到同一队列中并使用相同的池处理这些作业吗？

我想保持简洁，如果我创建一个新的API端点，我不必创建新的工作池，所以我可以只关注api。但是表现也非常考虑。

我试图构建的代码取自之前链接的示例，here是其他人使用此代码的github'gist'。

multithreading api go goroutine
3个回答
2
投票

前面的一件事：如果您正在运行HTTP服务器（无论如何都是Go的标准服务器），您无法在不停止和重新启动服务器的情况下控制goroutine的数量。每个请求至少启动一个goroutine，你无能为力。好消息是，这通常不是问题，因为goroutines非常轻巧。但是，你想要控制正在努力工作的goroutine的数量是完全合理的。

您可以将任何值放入通道，包括函数。因此，如果目标只是必须在http处理程序中编写代码，那么让工作关闭 - 工作人员不知道（或关心）他们正在做什么。

package main

import (
    "encoding/json"
    "io/ioutil"
    "net/http"
)

var largePool chan func()
var smallPool chan func()

func main() {
    // Start two different sized worker pools (e.g., for different workloads).
    // Cancelation and graceful shutdown omited for brevity.

    largePool = make(chan func(), 100)
    smallPool = make(chan func(), 10)

    for i := 0; i < 100; i++ {
            go func() {
                    for f := range largePool {
                            f()
                    }
            }()
    }

    for i := 0; i < 10; i++ {
            go func() {
                    for f := range smallPool {
                            f()
                    }
            }()
    }

    http.HandleFunc("/endpoint-1", handler1)
    http.HandleFunc("/endpoint-2", handler2) // naming things is hard, okay?

    http.ListenAndServe(":8080", nil)
}

func handler1(w http.ResponseWriter, r *http.Request) {
    // Imagine a JSON body containing a URL that we are expected to fetch.
    // Light work that doesn't consume many of *our* resources and can be done
    // in bulk, so we put in in the large pool.
    var job struct{ URL string }

    if err := json.NewDecoder(r.Body).Decode(&job); err != nil {
            http.Error(w, err.Error(), http.StatusBadRequest)
            return
    }

    go func() {
            largePool <- func() {
                    http.Get(job.URL)
                    // Do something with the response
            }
    }()

    w.WriteHeader(http.StatusAccepted)
}

func handler2(w http.ResponseWriter, r *http.Request) {
    // The request body is an image that we want to do some fancy processing
    // on. That's hard work; we don't want to do too many of them at once, so
    // so we put those jobs in the small pool.

    b, err := ioutil.ReadAll(r.Body)
    if err != nil {
            http.Error(w, err.Error(), http.StatusInternalServerError)
            return
    }

    go func() {
            smallPool <- func() {
                    processImage(b)
            }
    }()
    w.WriteHeader(http.StatusAccepted)
}

func processImage(b []byte) {}

这是一个非常简单的例子来说明问题。如何设置工作池并不重要。你只需要一个聪明的工作定义。在上面的示例中，它是一个闭包，但您也可以定义一个Job接口。

type Job interface {
    Do()
}

var largePool chan Job
var smallPool chan Job

现在，我不会将整个工作池方法称为“简单”。你说你的目标是限制goroutines（正在做的工作）的数量。这根本不需要工人;它只需要一个限制器。这是与上面相同的示例，但使用通道作为信号量来限制并发。

package main

import (
    "encoding/json"
    "io/ioutil"
    "net/http"
)

var largePool chan struct{}
var smallPool chan struct{}

func main() {
    largePool = make(chan struct{}, 100)
    smallPool = make(chan struct{}, 10)

    http.HandleFunc("/endpoint-1", handler1)
    http.HandleFunc("/endpoint-2", handler2)

    http.ListenAndServe(":8080", nil)
}

func handler1(w http.ResponseWriter, r *http.Request) {
    var job struct{ URL string }

    if err := json.NewDecoder(r.Body).Decode(&job); err != nil {
            http.Error(w, err.Error(), http.StatusBadRequest)
            return
    }

    go func() {
            // Block until there are fewer than cap(largePool) light-work
            // goroutines running.
            largePool <- struct{}{}
            defer func() { <-largePool }() // Let everyone that we are done

            http.Get(job.URL)
    }()

    w.WriteHeader(http.StatusAccepted)
}

func handler2(w http.ResponseWriter, r *http.Request) {
    b, err := ioutil.ReadAll(r.Body)
    if err != nil {
            http.Error(w, err.Error(), http.StatusInternalServerError)
            return
    }

    go func() {
            // Block until there are fewer than cap(smallPool) hard-work
            // goroutines running.
            smallPool <- struct{}{}
            defer func() { <-smallPool }() // Let everyone that we are done

            processImage(b)
    }()

    w.WriteHeader(http.StatusAccepted)
}

func processImage(b []byte) {}
0
投票

目前尚不清楚为什么你需要工人池？不会goroutines足够吗？

如果您受资源限制，可以考虑实施rates limiting。如果不是为什么根本不根据需要跨越惯例？

最好的学习方法是研究别人如何做好事。

看看https://github.com/valyala/fasthttp

Go的快速HTTP包。调整为高性能。热路径中的零内存分配。比net/http快10倍。

他们声称：

每个物理服务器提供超过150万个并发保持连接的高达200K rps的服务

这令人印象深刻，我怀疑你可以用pool / jobqueue做得更好。

0
投票

如前所述，在您的服务器中，每个请求处理程序将至少在一个goroutine中运行。

但是，如有必要，您仍然可以使用工作池来执行后端并行任务。例如，假设您的某些Http Handler函数触发对其他外部api的调用并将其结果“聚合”在一起，因此在这种情况下调用顺序无关紧要，这是一种可以利用工作池并分发您的工作，以使他们并行运行将每个任务分派给一个工人goroutine：

示例代码段：

    // build empty response
    capacity := config.GetIntProperty("defaultListCapacity")
    list := model.NewResponseList(make([]model.Response, 0, capacity), 1, 1, 0)

    // search providers
    providers := getProvidersByCountry(country)

    // create a slice of jobResult outputs
    jobOutputs := make([]<-chan job.JobResult, 0)

    // distribute work
    for i := 0; i < len(providers); i++ {
        job := search(providers[i], m)
        if job != nil {
            jobOutputs = append(jobOutputs, job.ReturnChannel)
            // Push each job onto the queue.
            GetInstance().JobQueue <- *job
        }
    }

    // Consume the merged output from all jobs
    out := job.Merge(jobOutputs...)
    for r := range out {
        if r.Error == nil {
            mergeSearchResponse(list, r.Value.(*model.ResponseList))
        }
    }
    return list

。工作池异步运行“通用”任务的完整示例：https://github.com/guilhebl/go-offer/blob/master/offer/repo.go

。使用的工作池库：https://github.com/guilhebl/go-worker-pool

最新问题
© www.soinside.com 2019 - 2024. All rights reserved.