Go高级特性之并发处理http详解

引言

在当今高度互联的世界中，Web 应用程序的性能和响应能力变得至关重要。大多数 Web 应用程序需要与多个外部服务进行通信，例如数据库、API、第三方服务等。并发发送 HTTP 请求是提高应用程序性能并保持响应能力的关键。Golang 作为一种高效的编程语言，提供了多种方法来实现并发发送 HTTP 请求。本文将深入探讨 Golang 中并发发送 HTTP 请求的最佳技术和实践。

使用 Goroutines 的基本方法

当谈到在 Golang 中实现并发时，最直接的方法是使用 routine。这些是 Go 中并发的构建块，提供了一种简单而强大的并发执行函数的方法。

Goroutine 入门

要启动一个 routine，只需在函数调用前加上``关键字即可。这会将函数作为 routine 启动，从而允许主程序继续独立运行。这就像开始一项任务并继续前进而不等待它完成。

例如，考虑发送 HTTP 请求的场景。通常，你会调用类似 的函数sendRequest()，并且你的程序将等待该函数完成。使用 routine，你可以同时执行此操作：

package main

import (
	"fmt"
	"net/http"
)

func main() {
	urls := []string{"http://example.com", "http://example.org"}

	for _, url := range urls {
		go func(url string) {
			resp, err := http.Get(url)
			if err != nil {
				fmt.Println("Error:", err)
				return
			}

			defer resp.Body.Close()

			// 处理响应内容或其他逻辑
		}(url)
	}

	// 等待所有请求完成
	// ...
}

这个循环为每个 URL 启动一个新的 routine，大大减少了程序发送所有请求所需的时间。通过创建多个 goroutine，每个 goroutine 发送一个 HTTP 请求，我们可以实现并发地发送多个请求。

并发 HTTP 请求的方法

由于没有限制 goroutine 数量，如果我们把全部任务都放到并发 Goroutine 中去执行，虽然效率比较高。但当不加控制的 goroutine 疯狂创建时候，服务器系统资源使用率飙升宕机，直到进程被自动 kill 不然无法提供任何其它服务。

上面的案例是我们不加控制 goroutine 数量限制从而导致宕机的，因此只要我们控制了 goroutine 数量就能避免这种问题！

WaitGroup

要确保所有并发请求完成后再继续执行后续操作，可以使用 sync.WaitGroup。sync 包提供的 WaitGroup 类型可以帮助我们等待一组并发操作的完成。

package main

import (
	"fmt"
	"net/http"
	"sync"
)

func main() {
	urls := []string{"http://example.com", "http://example.org"}

	var wg sync.WaitGroup

	for _, url := range urls {
		wg.Add(1)
		go func(url string) {
			defer wg.Done()

			resp, err := http.Get(url)
			if err != nil {
				fmt.Println("Error:", err)
				return
			}

			defer resp.Body.Close()

			// 处理响应内容或其他逻辑
		}(url)
	}

	wg.Wait() // 等待所有请求完成
	// ...
}

通过调用 sync.WaitGroup 的 Wait() 方法，我们可以确保所有的 goroutine 执行完成后再继续执行后续操作。

Channels

我们可以使用Channels来控制并发请求数量。通过创建一个有容量限制的通道，可以实现对并发请求数量的有效控制

package main

import (
	"fmt"
	"net/http"
)

func main() {
	urls := []string{"http://example.com", "http://example.org"}
	concurrency := 5
	sem := make(chan bool, concurrency)

	for _, url := range urls {
		sem <- true // 发送信号控制并发数量
		go func(url string) {
			defer func() { <-sem }() // 释放信号

			resp, err := http.Get(url)
			if err != nil {
				fmt.Println("Error:", err)
				return
			}

			defer resp.Body.Close()

			// 处理响应内容或其他逻辑
		}(url)
	}

	// 等待所有请求完成
	// ...
}

通过设置通道的容量为并发数量，我们可以确保同时发送的请求数量不超过设定的限制。这种方法对于控制并发请求数量非常有效。

Worker Pools

Worker Pools 是一种常见的并发模式，它可以有效地控制并发发送 HTTP 请求的数量。通过创建一组固定数量的 worker goroutines，并将请求任务分配给它们来处理，我们可以控制并发请求数量，减轻资源竞争和过载的压力。

package main

import (
	"fmt"
	"net/http"
	"sync"
)

type Worker struct {
	ID        int
	Request   chan string
	Responses chan string
}

func NewWorker(id int) *Worker {
	return &Worker{
		ID:        id,
		Request:   make(chan string),
		Responses: make(chan string),
	}
}

func (w *Worker) Start(wg *sync.WaitGroup) {
	go func() {
		defer wg.Done()

		for url := range w.Request {
			resp, err := http.Get(url)
			if err != nil {
				w.Responses <- fmt.Sprintf("Error: %s", err)
				continue
			}

			defer resp.Body.Close()

			// 处理响应内容或其他逻辑
			w.Responses <- fmt.Sprintf("Worker %d: %s", w.ID, "Response")
		}
	}()
}

func main() {
	urls := []string{"http://example.com", "http://example.org"}
	concurrency := 5

	var wg sync.WaitGroup
	wg.Add(concurrency)

	workers := make([]*Worker, concurrency)
	for i := 0; i < concurrency; i++ {
		workers[i] = NewWorker(i)
		workers[i].Start(&wg)
	}

	go func() {
		for _, url := range urls {
			for _, worker := range workers {
				worker.Request <- url
			}
		}

		for _, worker := range workers {
			close(worker.Request)
		}
	}()

	for _, worker := range workers {
		for response := range worker.Responses {
			fmt.Println(response)
		}
	}

	wg.Wait()
}

通过使用 Worker Pools，我们可以控制并发请求数量，并发请求的数量不超过 Worker Pools 中的 worker 数量。

更深入学习请查看探究 Go 的高级特性之 【处理1分钟百万请求】

使用信号量限制 Goroutines

sync/semaphore 包提供了一种干净有效的方法来限制并发运行的 routine 数量。当你想要更系统地管理资源分配时，此方法特别有用。

package main

import (
	"context"
	"fmt"
	"golang.org/x/sync/semaphore"
	"net/http"
)

func main() {
	// 创建请求者并加载配置
	requester := http.DefaultClient

	// 定义要处理的 URL 列表
	urls := []string{"http://example.com", "http://example.org", "http://example.net"}

	maxConcurrency := int64(2) // 设置最大并发请求数量

	// 创建一个带权重的信号量
	sem := semaphore.NewWeighted(maxConcurrency)

	ctx := context.Background()

	// 遍历 URL 列表
	for _, url := range urls {
		// 在启动 goroutine 前获取信号量权重
		if err := sem.Acquire(ctx, 1); err != nil {
			fmt.Printf("无法获取信号量：%v\n", err)
			continue
		}

		go func(url string) {
			defer sem.Release(1) // 在完成时释放信号量权重

			// 使用请求者获取 URL 对应的响应
			res, err := requester.Get(url)
			if err != nil {
				fmt.Printf("请求失败：%v\n", err)
				return
			}
			defer res.Body.Close()

			fmt.Printf("%s: %d\n", url, res.StatusCode)
		}(url)
	}

	// 等待所有 goroutine 释放它们的信号量权重
	if err := sem.Acquire(ctx, maxConcurrency); err != nil {
		fmt.Printf("等待时无法获取信号量：%v\n", err)
	}
}

那么，最好的方法是什么

在探索了 Go 中处理并发 HTTP 请求的各种方法之后，问题出现了：最好的方法是什么？正如软件工程中经常出现的情况一样，答案取决于应用程序的具体要求和约束。让我们考虑确定最合适方法的关键因素：

评估你的需求

• 请求规模：如果你正在处理大量请求，工作池或基于信号量的方法可以更好地控制资源使用。
• 错误处理：如果强大的错误处理至关重要，那么使用通道或信号量包可以提供更结构化的错误管理。
• 速率限制：对于需要遵守速率限制的应用程序，使用通道或信号量包限制 routine 可能是有效的。
• 复杂性和可维护性：考虑每种方法的复杂性。虽然渠道提供了更多控制，但它们也增加了复杂性。另一方面，信号量包提供了更直接的解决方案。

错误处理

由于 Go 中并发执行的性质，routines 中的错误处理是一个棘手的话题。由于 routine 独立运行，管理和传播错误可能具有挑战性，但对于构建健壮的应用程序至关重要。以下是一些有效处理并发 Go 程序中错误的策略：

集中误差通道

一种常见的方法是使用集中式错误通道，所有 routine 都可以通过该通道发送错误。然后，主 routine 可以监听该通道并采取适当的操作。

func worker(errChan chan<- error) {
    // 执行任务
    if err := doTask(); err != nil {
        errChan <- err // 将任何错误发送到错误通道
    }
}
func main() {
    errChan := make(chan error, 1) // 用于存储错误的缓冲通道
    go worker(errChan)
    if err := <-errChan; err != nil {
        // 处理错误
        log.Printf("发生错误：%v", err)
    }
}

或者你可以在不同的 routine 中监听 errChan。

func worker(errChan chan<- error, job Job) {
 // 执行任务
 if err := doTask(job); err != nil {
  errChan <- err // 将任何错误发送到错误通道
 }
}
func listenErrors(done chan struct{}, errChan <-chan error) {
 for {
  select {
  case err := <-errChan:
   // 处理错误
  case <-done:
   return
  }
 }
}
func main() {
 errChan := make(chan error, 1000) // 存储错误的通道
 done := make(chan struct{})       // 用于通知 goroutine 停止的通道
 go listenErrors(done, errChan)
 
 for _, job := range jobs {
   go worker(errChan, job)
 }
 // 等待所有 goroutine 完成（具体方式需要根据代码的实际情况进行实现）
 done <- struct{}{} // 通知 goroutine 停止监听错误
}

Error Group

lang.org/x/sync/errgroup 包提供了一种便捷的方法来对多个 routine 进行分组并处理它们产生的任何错误。errgroup.Group确保一旦任何 routine 发生错误，所有后续操作都将被取消。

import "golang.org/x/sync/errgroup"
func main() {
    g, ctx := errgroup.WithContext(context.Background())
    urls := []string{"http://example.com", "http://example.org"}
    for _, url := range urls {
        // 为每个 URL 启动一个 goroutine
        g.Go(func() error {
            // 替换为实际的 HTTP 请求逻辑
            _, err := fetchURL(ctx, url)
            return err
        })
    }
    // 等待所有请求完成
    if err := g.Wait(); err != nil {
        log.Printf("发生错误：%v", err)
    }
}

这种方法简化了错误处理，特别是在处理大量 routine 时。

以上就是Go高级特性之并发处理http详解的详细内容，更多关于Go并发处理http的资料请关注其它相关文章！