快捷導(dǎo)航

Go語(yǔ)言中進(jìn)行API限流的實(shí)戰(zhàn)詳解

更新時(shí)間：2025年01月24日 10:11:50 作者：Ai?編碼

API?限流是控制和管理應(yīng)用程序訪問(wèn)量的重要手段,旨在防止惡意濫用、保護(hù)后端服務(wù)的穩(wěn)定性和可用性,下面我們就來(lái)看看如何在Go語(yǔ)言中具體實(shí)現(xiàn)吧

為什么要進(jìn)行 API 限流

API 限流是控制和管理應(yīng)用程序訪問(wèn)量的重要手段，旨在防止惡意濫用、保護(hù)后端服務(wù)的穩(wěn)定性和可用性，并確保系統(tǒng)能夠有效處理請(qǐng)求。API 限流不僅能提高系統(tǒng)的性能，還能優(yōu)化用戶體驗(yàn)。下面是為什么要進(jìn)行 API 限流的一些原因：

1.防止服務(wù)過(guò)載

當(dāng) API 接口的訪問(wèn)量過(guò)大時(shí)，后端服務(wù)可能無(wú)法處理所有請(qǐng)求，導(dǎo)致服務(wù)崩潰或響應(yīng)遲緩。

限流能夠保證后端服務(wù)不會(huì)因?yàn)楦卟l(fā)請(qǐng)求而被擊垮，從而提升系統(tǒng)的可靠性。

2.保護(hù)資源

API 服務(wù)通常需要訪問(wèn)數(shù)據(jù)庫(kù)、緩存、第三方 API 等資源。通過(guò)限流，可以控制這些資源的訪問(wèn)頻率，防止它們被過(guò)度消耗。

防止用戶發(fā)送大量無(wú)效請(qǐng)求或重復(fù)請(qǐng)求，避免不必要的資源浪費(fèi)。

3.提高用戶體驗(yàn)

API 限流能夠確保每個(gè)用戶公平地獲得訪問(wèn)權(quán)限，避免某些用戶發(fā)送過(guò)多請(qǐng)求從而影響其他用戶的體驗(yàn)。

4.防止惡意攻擊

限流是抵御 DDoS（分布式拒絕服務(wù)）攻擊的一種手段，可以有效減緩大量請(qǐng)求的沖擊。

防止暴力破解（如密碼暴力破解、賬戶濫用等）攻擊，限制某些頻繁請(qǐng)求的行為。

5.成本控制

許多服務(wù)（如云服務(wù)、第三方 API）基于請(qǐng)求量計(jì)費(fèi)。通過(guò)限流，能夠在一定程度上減少不必要的成本。

常用的 API 限流算法

以下是一些常見(jiàn)的 API 限流算法，每種算法有不同的優(yōu)缺點(diǎn)和適用場(chǎng)景：

1. 令牌桶算法（Token Bucket）

原理：令牌桶算法維護(hù)一個(gè)桶，每個(gè)請(qǐng)求需要獲取一個(gè)令牌。令牌會(huì)以固定速率放入桶中，桶有最大容量。如果桶滿了，新的令牌將會(huì)丟棄。當(dāng)請(qǐng)求到達(dá)時(shí)，若桶中有令牌，說(shuō)明可以處理該請(qǐng)求，令牌被取走；若桶中沒(méi)有令牌，則拒絕請(qǐng)求。

特點(diǎn)：

能夠處理突發(fā)流量，因?yàn)橥爸锌梢苑e累一定數(shù)量的令牌。
適用于允許請(qǐng)求量有一定波動(dòng)的場(chǎng)景。

適用場(chǎng)景：適用于限流場(chǎng)景中需要允許短時(shí)間內(nèi)的流量突增，例如某些高并發(fā)的 API 服務(wù)。

2. 漏桶算法（Leaky Bucket）

原理：漏桶算法也使用一個(gè)桶來(lái)存儲(chǔ)請(qǐng)求，當(dāng)請(qǐng)求到達(dá)時(shí)，它們會(huì)依次進(jìn)入桶中。如果桶沒(méi)有滿，水會(huì)以恒定的速率流出。如果桶已滿，則新請(qǐng)求會(huì)被丟棄。

特點(diǎn)：

平滑的流量控制，能確保請(qǐng)求按照固定速率流出。
不允許短時(shí)間內(nèi)出現(xiàn)突發(fā)流量，所有的請(qǐng)求流量必須按固定速率流出。

適用場(chǎng)景：適用于需要平滑流量、避免短時(shí)間內(nèi)的突發(fā)請(qǐng)求影響系統(tǒng)的場(chǎng)景。

3. 固定窗口計(jì)數(shù)算法（Fixed Window Counter）

原理：在固定的時(shí)間窗口內(nèi)（如每分鐘或每小時(shí)），允許一定數(shù)量的請(qǐng)求通過(guò)。每當(dāng)時(shí)間窗口結(jié)束時(shí)，計(jì)數(shù)器會(huì)重置。超出最大請(qǐng)求數(shù)的請(qǐng)求將被拒絕。

特點(diǎn)：

簡(jiǎn)單易實(shí)現(xiàn)，但容易受到時(shí)間窗口邊界問(wèn)題的影響（即在窗口切換時(shí)，可能會(huì)允許過(guò)多的請(qǐng)求通過(guò)）。

適用場(chǎng)景：適用于流量相對(duì)平穩(wěn)的場(chǎng)景，通常在請(qǐng)求量較低的應(yīng)用中使用。

4. 滑動(dòng)窗口計(jì)數(shù)算法（Sliding Window Counter）

原理：與固定窗口計(jì)數(shù)算法類似，但是它不是固定的時(shí)間窗口，而是以滑動(dòng)的方式進(jìn)行計(jì)數(shù)。每次請(qǐng)求都會(huì)在時(shí)間軸上記錄，滑動(dòng)窗口會(huì)隨著時(shí)間的推移進(jìn)行更新。

特點(diǎn)：

能夠更加平滑地控制請(qǐng)求速率，避免固定窗口切換時(shí)的突發(fā)流量。
適用場(chǎng)景：適用于需要精確限流的場(chǎng)景，減少邊界效應(yīng)。

5. 基于 Redis 的限流

原理：使用 Redis 的鍵值存儲(chǔ)特性，可以結(jié)合 Redis 的過(guò)期時(shí)間來(lái)實(shí)現(xiàn)限流。例如，在 Redis 中記錄每個(gè)用戶或 IP 的請(qǐng)求次數(shù)，并設(shè)置過(guò)期時(shí)間，來(lái)實(shí)現(xiàn)請(qǐng)求的限制。

特點(diǎn)：

分布式限流，非常適合分布式系統(tǒng)。
高效、易擴(kuò)展。

適用場(chǎng)景：適用于分布式系統(tǒng)，尤其是微服務(wù)架構(gòu)中的 API 限流。

使用 Go 實(shí)現(xiàn) API 限流

以下是如何在 Go 中實(shí)現(xiàn) API 限流的幾種常見(jiàn)方式：

1. 基于 Token Bucket 算法實(shí)現(xiàn)限流

package main

import (
	"fmt"
	"time"
)

type TokenBucket struct {
	capacity int           // 桶的最大容量
	tokens   int           // 當(dāng)前桶中令牌數(shù)量
	rate     time.Duration // 令牌生成速率
	lastTime time.Time     // 上次令牌生成時(shí)間
}

func NewTokenBucket(capacity int, rate time.Duration) *TokenBucket {
	return &TokenBucket{
		capacity: capacity,
		tokens:   capacity,
		rate:     rate,
		lastTime: time.Now(),
	}
}

func (tb *TokenBucket) Allow() bool {
	// 計(jì)算令牌的生成數(shù)量
	now := time.Now()
	elapsed := now.Sub(tb.lastTime)
	tokensToAdd := int(elapsed / tb.rate)
	tb.lastTime = now

	// 如果有令牌生成，更新桶中的令牌數(shù)量
	if tokensToAdd > 0 {
		tb.tokens = min(tb.capacity, tb.tokens+tokensToAdd)
	}

	// 如果有令牌，則消費(fèi)一個(gè)令牌并返回 true
	if tb.tokens > 0 {
		tb.tokens--
		return true
	}
	// 否則返回 false，拒絕請(qǐng)求
	return false
}

func min(a, b int) int {
	if a < b {
		return a
	}
	return b
}

func main() {
	tb := NewTokenBucket(5, time.Second)

	// 模擬請(qǐng)求
	for i := 0; i < 10; i++ {
		if tb.Allow() {
			fmt.Println("Request", i, "allowed")
		} else {
			fmt.Println("Request", i, "denied")
		}
		time.Sleep(200 * time.Millisecond)
	}
}

2. 基于 Redis 實(shí)現(xiàn)限流（使用 Go Redis 客戶端）

使用 Redis 進(jìn)行限流實(shí)現(xiàn)時(shí)，常用的是 SETNX（設(shè)置一個(gè)鍵值對(duì)，如果鍵不存在則設(shè)置，防止重復(fù)計(jì)數(shù)）和 EXPIRE（設(shè)置過(guò)期時(shí)間）來(lái)控制訪問(wèn)頻率。

package main

import (
	"fmt"
	"log"
	"time"

	"github.com/go-redis/redis/v8"
	"golang.org/x/net/context"
)

var rdb *redis.Client
var ctx = context.Background()

func init() {
	// 初始化 Redis 客戶端
	rdb = redis.NewClient(&redis.Options{
		Addr: "localhost:6379", // Redis 地址
	})
}

func limitRequest(userID string) bool {
	// 使用 Redis 來(lái)存儲(chǔ)訪問(wèn)記錄
	key := fmt.Sprintf("rate_limit:%s", userID)
	// 設(shè)置每個(gè)用戶的最大請(qǐng)求次數(shù)限制
	maxRequests := 5
	// 設(shè)置過(guò)期時(shí)間為 60 秒
	expiration := 60 * time.Second

	// 獲取當(dāng)前用戶的請(qǐng)求次數(shù)
	count, err := rdb.Get(ctx, key).Int()
	if err != nil && err != redis.Nil {
		log.Fatalf("Error retrieving count from Redis: %v", err)
	}

	// 如果請(qǐng)求次數(shù)超過(guò)限制，拒絕請(qǐng)求
	if count >= maxRequests {
		return false
	}

	// 增加請(qǐng)求次數(shù)
	err = rdb.Incr(ctx, key).Err()
	if err != nil {
		log.Fatalf("Error incrementing count: %v", err)
	}

	// 設(shè)置過(guò)期時(shí)間，避免無(wú)限制的請(qǐng)求
	rdb.Expire(ctx, key, expiration)

	return true
}

func main() {
	userID := "user123"
	for i := 0; i < 10; i++ {
		if limitRequest(userID) {
			fmt.Println("Request allowed")
		} else {
			fmt.Println("Request denied due to rate limit")
		}
		time.Sleep(5 * time.Second)
	}
}