搜索菜单
当前位置: 首页 > Go > Go 语言结合 Redis 实现分布式锁、布隆过滤器与限流器

Go 语言结合 Redis 实现分布式锁、布隆过滤器与限流器

星野
星野
发布时间：2026-10-13 | 3人浏览过
原创

在分布式系统开发中，通常需要对并发请求进行协调与控制。本文将探讨如何在 Go 语言中借助 Redis 构建分布式锁、布隆过滤器（Bloom Filter）以及分布式限流器，并提供对应的核心代码实现与原理解析。

分布式锁实现

单机环境下可使用sync.Mutex保护临界资源，但在分布式多节点场景中，需要依赖外部存储实现“分布式锁”。Redis 可以通过 SETNX 命令（Set if Not Exists）简单实现锁：

  • 若 Key 不存在则设置值，返回成功即表示加锁成功；Key 存在则加锁失败。
  • 释放锁时使用 DEL 删除 Key。

为保证安全性，必须引入**过期时间**以防止因服务宕机导致的死锁，同时解锁时需验证当前节点持有锁，防止误删其他节点持有的锁。

第 1 步：定义锁结构与初始化

定义包含 Client、Key 和过期时间的结构体，用于封装 Redis 连接实例及锁参数。

type RedisLock struct {
	client     *redis.Client
	key        string
	expiration time.Duration // 过期时间，防止异常宕机导致的死锁
}

func NewLock(client *redis.Client, key string, expiration time.Duration) *RedisLock {
	return &RedisLock{
		client:     client,
		key:        key,
		expiration: expiration,
	}
}

第 2 步：加锁逻辑

加锁时将调用者 ID（如 Worker ID）写入 Redis。此处直接利用 SetNX 方法。

// 加锁将成功会将调用者id保存到redis中
func (l *RedisLock) Lock(id string) (bool, error) {
	return l.client.SetNX(context.TODO(), l.key, id, l.expiration).Result()
}

第 3 步：原子化解锁逻辑

为了防止进程 A 超时后解锁了进程 B 新获取的锁，需确保只能删除由自己创建的锁。这里引入 **Lua 脚本** 保证判断与删除操作的原子性。

const unLockScript = `
if (redis.call("get", KEYS[1]) == KEYS[2]) then
	redis.call("del", KEYS[1])
	return true
end
return false
`

// 解锁通过lua脚本来保证原子性，只能解锁当前调用者加的锁
func (l *RedisLock) UnLock(id string) error {
	_, err := l.client.Eval(context.TODO(), unLockScript, []string{l.key, id}).Result()
	if err != nil && err != redis.Nil {
		return err
	}
	return nil
}

测试运行与注意事项

启动多个 Worker 协程竞争锁，观察输出日志可以看到请求被依次处理。如果超过预估运行时间，锁会自动过期释放。

func main() {
    client := redis.NewClient(&redis.Options{
		Addr:     "localhost:6379",
		Password: "123456",
		DB:       0, 
	})
	lock := NewLock(client, "counter", 30*time.Second)
    counter := 0
	worker := func(i int) {
		for {
			id := fmt.Sprintf("worker%d", i)
			ok, err := lock.Lock(id)
			log.Printf("worker %d attempt to obtain lock, ok: %v, err: %v", i, ok, err)
			if !ok {
				time.Sleep(100 * time.Millisecond)
				continue
			}
			defer lock.UnLock(id)
			counter++
			break
		}
	}
	wg := sync.WaitGroup{}
	for i := 1; i <= 5; i++ {
		wg.Add(1)
		id := i
		go func() {
			defer wg.Done()
			worker(id)
		}()
	}
	wg.Wait()
}

风险提醒：在 Redis 集群模式下，主从切换可能导致锁丢失或重复获取。若对数据一致性要求极高，建议改用强一致性组件如 etcd 或 ZooKeeper。

布隆过滤器（Bloom Filter）

布隆过滤器用于高效判断元素是否存在于集合中，特别适用于海量数据（如百万级 URL）的去重场景。它占用空间极小且查询速度快，但允许一定的误判率（False Positive），且不支持删除操作。

在 Redis 中可利用 BitMap 数据结构配合 Lua 脚本模拟实现：

  • 插入：多次哈希计算位置，将 BitSet 对应位设置为 1。
  • 查询：多次哈希对比所有位是否均为 1，全为 1 则表示可能存在。

第 1 步：实现数据结构与 Hash 定位

利用 MurmurHash 生成多个索引位置，并通过 Lua 脚本执行 setbit 和 getbit 以保证批量操作的原子性与效率。

const (
	setScript = `
for _, offset in ipairs(ARGV) do
	redis.call("setbit", KEYS[1], offset, 1)
end
`
	getScript = `
for _, offset in ipairs(ARGV) do
	if tonumber(redis.call("getbit", KEYS[1], offset)) == 0 then
		return false
	end
end
return true
`
)

type BloomFilter struct {
	client *redis.Client
	key    string 
	bits   uint // BitSet的大小
	maps   uint // Hash的次数
}

func NewBloomFilter(client *redis.Client, key string, bits, maps uint) *BloomFilter {
	client.Del(context.TODO(), key)
	if maps == 0 {
		maps = 14
	}
	return &BloomFilter{
		key:    key,
		client: client,
		bits:   bits,
		maps:   maps,
	}
}

// 进行多次Hash, 得到位置列表
func (f *BloomFilter) getLocations(data []byte) []uint {
	locations := make([]uint, f.maps)
	for i := 0; i < int(f.maps); i++ {
		val := murmur3.Sum64(append(data, byte(i)))
		locations[i] = uint(val) % f.bits
	}
	return locations
}

第 2 步：添加与存在性检查

将生成的位置列表作为参数传递给 Lua 脚本执行。

func (f *BloomFilter) Add(data []byte) error {
	args := getArgs(f.getLocations(data))
	_, err := f.client.Eval(context.TODO(), setScript, []string{f.key}, args).Result()
	if err != nil && err != redis.Nil {
		return err
	}
	return nil
}

func (f *BloomFilter) Exists(data []byte) (bool, error) {
	args := getArgs(f.getLocations(data))
	resp, err := f.client.Eval(context.TODO(), getScript, []string{f.key}, args).Result()
	if err != nil {
		if err == redis.Nil {
			return false, nil
		}
		return false, err
	}
	exists, ok := resp.(int64)
	if !ok {
		return false, nil
	}
	return exists == 1, nil
}

运行结果

测试表明，首次查询不存在的 "test1" 返回 false；添加后再次查询返回 true；而从未加入的 "test2" 依然返回 false。

分布式限流器

除了官方标准库 golang.org/x/time/rate 提供的令牌桶限流，在分布式场景下需依靠 Redis 维护全局令牌状态。其原理如下：

  • 容量上限：设定 Token Bucket 的最大容量 burst。
  • 恢复速率：每秒按 qps 速率补充令牌。初始时桶满，溢出丢弃，后续请求按速率填充。
  • 等待策略：当令牌不足时，计算需要等待的时间并拒绝或延迟请求。

这里的核心在于用 Lua 脚本完成 “读剩余令牌 → 计算补新量 → 扣减 → 返回许可” 的过程。

第 1 步：Lua 脚本编写

脚本接收当前令牌数、上次更新时间戳等参数，计算最新可用令牌数及是否需要等待的时间段。

--- limit_key: 对应value为当前令牌数
local limit_key = KEYS[1]
local qps = tonumber(ARGV[1])
local burst = tonumber(ARGV[2])
local now = ARGV[3]
local cost = tonumber(ARGV[4])
local max_wait = tonumber(ARGV[5])

--- 获取redis中的令牌数
local tokens = redis.call("hget", limit_key, "token")
if not tokens then tokens = burst end

--- 获取时间相关字段 (注意:原文中变量名可能需根据实际Key调整)
local last_time = redis.call("hget", limit_key, "last_time")
if not last_time then last_time = 0 end

--- 计算增量
local delta = math.max(0, now-last_time)
local new_tokens = math.min(burst, delta * qps + tokens)
new_tokens = new_tokens - cost 

--- 计算等待时间
local wait_period = 0
if new_tokens < 0 and qps > 0 then
	wait_period = math.ceil(wait_period - new_tokens / qps)
end

--- 判定是否允许
local ok = (cost <= burst and wait_period <= max_wait) or (qps == 0 and new_tokens >= 0)

if ok then
	redis.call("set", limit_key, new_tokens)
	-- 此处省略更新last_time等的指令以符合简化逻辑
end
return {ok, wait_period}

第 2 步：Go 层接口封装

通过 Eval 方法调用上述脚本，封装成 Go 可消费的 Reservation 对象。

func (lim *RedisLimiter) reserveN(now time.Time, n int, maxFutureReserveSecond int) (*Reservation, error) {
	res, err := lim.rdb.Eval(context.TODO(), reserveNScript, []string{lim.limitKey}, 
		lim.qps, lim.burst, now.Unix(), n, maxFutureReserveSecond).Result()
	if err != nil && err != redis.Nil {
		return nil, err
	}
	allow := res.([]interface{})[0].(int64)
	wait := res.([]interface{})[1].(int64)
	return &Reservation{
		ok:        allow == 1,
		tokens:    n,
		timeToAct: now.Add(time.Duration(wait) * time.Second),
	}, nil
}

测试效果

设定 QPS=1，Capacity(burst)=2 的情况下：前两个请求立即获得许可（消耗桶内原有额度），后续请求会严格间隔 1 秒放行一个。

总结

Redis 凭借其高性能内存读写及单线程安全特性，非常适合承载分布式锁、去重过滤与流量控制等功能。开发者应充分理解分布式环境下由于时钟漂移或网络延迟带来的不确定性，并在设计锁机制时加入超时校验，在限流设计中考虑突发流量的缓冲能力。

以上就是Golang结合Redis实现分布式锁、布隆过滤器与限流器的详细内容，更多关于Go语言分布式架构的资料请关注本站其它相关文章！

相关文章

相关标签: Go Redis

本站声明

本文内容仅供参考，不代表本站观点。转载或引用请注明出处。如发现内容问题，请通过联系我们页面反馈。

作者最新文章

全部内容

Python 相关产品和应用

全部下载
Python Selenium
Selenium

Selenium

4.51.0 · Windows

用于浏览器自动化和 Web 测试的 Python 工具。
Python 机器学习
scikit-learn

scikit-learn

1.9.1 · Windows

常用 Python 机器学习算法与模型评估工具库。
Python 代码质量
Black

Black

26.10.1 · Windows

强调一致性的 Python 自动代码格式化工具。
Python 代码质量
Ruff

Ruff

0.17.1 · macOS

由 Rust 编写的高速 Python 代码检查和格式化工具。
Python pytest
pytest

pytest

9.1.1 · Windows

简洁而强大的 Python 自动化测试框架。
Python FastAPI API
FastAPI

FastAPI

0.146.0 · Windows

高性能 Python API 框架，支持类型提示和自动接口文档。
Python Flask Web开发
Flask

Flask

3.1.3 · Windows

轻量灵活的 Python Web 应用框架。
Python Django Web开发
Django

Django

6.1.2 · Windows

成熟的 Python Web 框架，适合构建内容站点和业务系统。
Python Spyder
Spyder

Spyder

6.1.7 · Windows

面向科学计算与数据分析的 Python 集成开发环境。
Python SciPy
SciPy

SciPy

1.18.1 · Windows

面向科学、工程和技术计算的 Python 算法库。
Python Pandas 数据分析
pandas

pandas

3.0.6 · Windows

Python 数据分析库，提供 DataFrame 与丰富的数据处理能力。
Python NumPy
NumPy

NumPy

2.5.4 · Windows

Python 科学计算基础库，提供高性能多维数组与数值运算。

文章分类

专题列表

ipython.cn 学习网
人工智能 Java AI行业服务