집 > 기사 > 데이터 베이스 > Golang 분산 애플리케이션에서 Redis를 사용하는 방법

Golang 분산 애플리케이션에서 Redis를 사용하는 방법

王林앞으로: 2023-05-26 22:07:36846검색

Text

Redis는 분산 시스템에서 자주 사용되는 고성능 인 메모리 데이터베이스이며, 분산 캐시 또는 간단한 인 메모리 데이터베이스 외에도 몇 가지 특별한 애플리케이션 시나리오가 있습니다. Golang 해당 미들웨어와 결합하여 작성되었습니다.

분산 잠금

단일 시스템에서는 sync.Mutex를 사용하여 중요한 리소스를 보호할 수 있습니다. 분산 시스템에도 이러한 요구가 있습니다. 해당 "분산 잠금"입니다. sync.Mutex来保护临界资源，在分布式系统中同样有这样的需求，当多个主机抢占同一个资源，需要加对应的“分布式锁”。

在Redis中我们可以通过setnx命令来实现

如果key不存在可以设置对应的值，设置成功则加锁成功，key不存在返回失败
释放锁可以通过del实现。

主要逻辑如下：

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

为了防止系统宕机或异常请求导致的死锁，需要添加一个额外的超时时间，该超时时间应设为最大估计运行时间的两倍。

解锁时通过lua脚本来保证原子性，调用者只会解自己加的锁。避免由于超时造成的混乱，例如：进程A在时间t1获取了锁，但由于执行缓慢，在时间t2锁超时失效，进程B在t3获取了锁，这是如果进程A执行完去解锁会取消进程B的锁。

运行测试

func main() {
    client := redis.NewClient(&redis.Options{
		Addr:     "localhost:6379",
		Password: "123456",
		DB:       0, // use default DB
	})
	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++
			log.Printf("worker %d, add counter %d", i, 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()
}

运行结果，可以看到与sync.Mutex使用效果类似

2022/07/22 09:58:09 worker 5 attempt to obtain lock, ok: true, err: 2d77b2345c34a631c3d251f57ce68620
2022/07/22 09:58:09 worker 5, add counter 1
2022/07/22 09:58:09 worker 4 attempt to obtain lock, ok: false, err: 2d77b2345c34a631c3d251f57ce68620
2022/07/22 09:58:09 worker 1 attempt to obtain lock, ok: false, err: 2d77b2345c34a631c3d251f57ce68620
2022/07/22 09:58:09 worker 2 attempt to obtain lock, ok: false, err: 2d77b2345c34a631c3d251f57ce68620
2022/07/22 09:58:09 worker 3 attempt to obtain lock, ok: false, err: 2d77b2345c34a631c3d251f57ce68620
2022/07/22 09:58:10 worker 3 attempt to obtain lock, ok: false, err: 2d77b2345c34a631c3d251f57ce68620
2022/07/22 09:58:10 worker 1 attempt to obtain lock, ok: false, err: 2d77b2345c34a631c3d251f57ce68620
2022/07/22 09:58:10 worker 2 attempt to obtain lock, ok: false, err: 2d77b2345c34a631c3d251f57ce68620
2022/07/22 09:58:10 worker 4 attempt to obtain lock, ok: true, err: 2d77b2345c34a631c3d251f57ce68620
2022/07/22 09:58:10 worker 4, add counter 2
2022/07/22 09:58:10 worker 1 attempt to obtain lock, ok: true, err: 2d77b2345c34a631c3d251f57ce68620
2022/07/22 09:58:10 worker 1, add counter 3
2022/07/22 09:58:10 worker 3 attempt to obtain lock, ok: false, err: 2d77b2345c34a631c3d251f57ce68620
2022/07/22 09:58:10 worker 2 attempt to obtain lock, ok: false, err: 2d77b2345c34a631c3d251f57ce68620
2022/07/22 09:58:10 worker 2 attempt to obtain lock, ok: true, err: 2d77b2345c34a631c3d251f57ce68620
2022/07/22 09:58:10 worker 2, add counter 4
2022/07/22 09:58:10 worker 3 attempt to obtain lock, ok: false, err: 2d77b2345c34a631c3d251f57ce68620
2022/07/22 09:58:10 worker 3 attempt to obtain lock, ok: true, err: 2d77b2345c34a631c3d251f57ce68620
2022/07/22 09:58:10 worker 3, add counter 5

特别注意的是，在分布式Redis集群中，如果发生异常时(主节点宕机)，可能会降低分布式锁的可用性，可以通过强一致性的组件etcd、ZooKeeper等实现。

分布式过滤器

假设要开发一个爬虫服务，爬取百万级的网页，怎么判断某一个网页是否爬取过，除了借助数据库和HashMap，我们可以借助布隆过滤器来做。相对于其他方法，布隆过滤器占用空间非常少，且插入和查询时间非常快。

布隆过滤器用来判断某个元素是否在集合中，利用BitSet

插入数据时将值进行多次Hash，将BitSet对应位置1
查询时同样进行多次Hash对比所有位上是否为1，如是则存在。

布隆过滤器有一定的误判率，不适合精确查询的场景。另外也不支持删除元素。通常适用于URL去重、垃圾邮件过滤、防止缓存击穿等场景中。

在Redis中，我们可以使用自带的BitSet实现，同样也借助lua脚本的原子性来避免多次查询数据不一致。

const (
	// 插入数据，调用setbit设置对应位
	setScript = `
for _, offset in ipairs(ARGV) do
	redis.call("setbit", KEYS[1], offset, 1)
end
`
	// 查询数据，如果所有位都为1返回true
	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 // 存在redis中的key
	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
}
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
}
func getArgs(locations []uint) []string {
	args := make([]string, 0)
	for _, l := range locations {
		args = append(args, strconv.FormatUint(uint64(l), 10))
	}
	return args
}

运行测试

func main() {
	bf := NewBloomFilter(client,"bf-test", 2^16, 14)
	exists, err := bf.Exists([]byte("test1"))
	log.Printf("exist %t, err %v", exists, err)
	if err := bf.Add([]byte("test1")); err != nil {
		log.Printf("add err: %v", err)
	}
	exists, err = bf.Exists([]byte("test1"))
	log.Printf("exist %t, err %v", exists, err)
	exists, err = bf.Exists([]byte("test2"))
	log.Printf("exist %t, err %v", exists, err)
// output
// 2022/07/22 10:05:58 exist false, err <nil>
// 2022/07/22 10:05:58 exist true, err <nil>
// 2022/07/22 10:05:58 exist false, err <nil>
}

分布式限流器

在golang.org/x/time/rate

Redis에서는 setnx 명령을 통해 이를 달성할 수 있습니다

키가 없으면 해당 값을 설정할 수 있습니다. 설정이 성공하면 잠금이 성공합니다. 키가 존재하지 않으면 실패를 반환합니다
릴리스 잠금은 del을 통해 구현할 수 있습니다.
주요 로직은 다음과 같습니다.
```
--- 相关Key
--- limit rate key值，对应value为当前令牌数
local limit_key = KEYS[1]
--- 输入参数
--[[
qps: 每秒请求数;
burst: 令牌桶容量;
now: 当前Timestamp;
cost: 请求令牌数;
max_wait: 最大等待时间
--]]
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
--- 上次修改时间
local last_time = redis.call("hget", limit_key, "last_time")
if not last_time then
	last_time = 0
end
--- 最新等待时间
local last_event = redis.call("hget", limit_key, "last_event")
if not last_event then
	last_event = 0
end
--- 通过当前时间与上次修改时间的差值，qps计算出当前时间得令牌数
local delta = math.max(0, now-last_time)
local new_tokens = math.min(burst, delta * qps + tokens)
new_tokens = new_tokens - cost --- 最新令牌数，减少请求令牌
--- 如果最新令牌数小于0，计算需要等待的时间
local wait_period = 0
if new_tokens < 0 and qps > 0 then
	wait_period = wait_period - new_tokens / qps
end
wait_period = math.ceil(wait_period)
local time_act = now + wait_period --- 满足等待间隔的时间戳
--- 允许请求有两种情况
--- 当请求令牌数小于burst, 等待时间不超过最大等待时间，可以通过补充令牌满足请求
--- qps为0时，只要最新令牌数不小于0即可
local ok = (cost <= burst and wait_period <= max_wait and qps > 0) or (qps == 0 and new_tokens >= 0)
--- 设置对应值
if ok then
	redis.call("set", limit_key, new_tokens)
	redis.call("set", last_time_key, now)
	redis.call("set", last_event_key, time_act)
end
--- 返回列表，{是否允许， 等待时间}
return {ok, wait_period}
```
시스템 다운타임이나 비정상적인 요청으로 인한 교착 상태를 방지하기 위해 추가 시간 초과를 추가해야 하며, 이 시간은 최대 예상 실행 시간의 두 배로 설정해야 합니다. 🎜🎜잠금 해제 시 원자성을 보장하려면 Lua 스크립트를 사용하세요. 호출자는 자체적으로 추가된 잠금만 잠금 해제합니다. 시간 초과로 인한 혼란 방지 예: 프로세스 A가 시간 t1에 잠금을 획득했지만 느린 실행으로 인해 시간 t2에 잠금 시간이 초과되었습니다. 프로세스 B가 t3에 잠금을 획득한 경우 B의 잠금이 취소됩니다. 🎜
테스트 실행
```
// 调用lua脚本
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
	}
	//...
	return &Reservation{
		ok:        allow == 1,
		lim:       lim,
		tokens:    n,
		timeToAct: now.Add(time.Duration(wait) * time.Second),
	}, nil
}
```
🎜실행 결과 sync.Mutex를 사용한 것과 효과가 비슷한 것을 확인할 수 있습니다🎜
🎜2022/07/22 09 :58:09 작업자 5가 잠금을 얻으려고 시도함, ok: true, err:
2022/07/22 09:58:09 작업자 5, 카운터 1 추가
2022/07/22 09:58:09 작업자 4 잠금 획득 시도, ok: false, err:
2022/07/22 09:58:09 작업자 1 잠금 획득 시도, ok: false, err: < ;nil>
2022/07/22 09:58:09 작업자 2가 잠금을 얻으려고 시도했습니다. ok: false, err: 2d77b2345c34a631c3d251f57ce68620
2022/07/22 09:58:09 작업자 3 잠금 획득 시도, ok: false, err:
2022/07/22 09:58:10 작업자 3 잠금 획득 시도, ok: false, err:
2022/07/22 09: 58:10 작업자 1이 잠금을 얻으려는 시도, ok: false, err:
2022/07/22 09:58:10 작업자 2가 잠금을 얻으려고 시도, ok: false, 오류: 2d77b2345c34a631c3d251f57ce68620
2022/07/22 09:58:10 작업자 4가 잠금을 얻으려고 시도했습니다. ok: true, 오류: 2d77b2345c34a631c3d251f57ce68620
2022/07/22 09: 58:10 작업자 4, 카운터 2 추가
2022/07/22 09:58:10 작업자 1 잠금 획득 시도, ok: true, err:
2022/07/22 09 :58:10 작업자 1, 카운터 3 추가
2022/07/22 09:58:10 작업자 3이 잠금을 얻으려고 시도, ok: false, err: 2d77b2345c34a631c3d251f57ce68620
2022/07/22 09:58:10 작업자 2가 잠금을 얻으려고 시도, ok: false, err:
2022/07/22 09:58:10 작업자 2가 잠금을 얻으려고 시도, ok: true, err: < ;nil>
2022/07/22 09:58:10 작업자 2, 카운터 4 추가
2022/07/22 09:58:10 작업자 3이 잠금을 얻으려고 시도합니다. ok: false, err:
2022/07/22 09:58:10 작업자 3이 잠금을 얻으려고 시도했습니다. ok: true, err:
2022/07/22 09:58:10 작업자 3, 카운터 5 추가🎜🎜분산 Redis 클러스터에서 예외가 발생하면(마스터 노드가 다운됨) 분산 잠금의 가용성이 줄어들 수 있다는 점에 특별한 주의가 필요합니다. 이는 강력한 일관성 구성 요소를 통해 달성할 수 있습니다. etcd, ZooKeeper 등이 있습니다. 🎜🎜분산 필터🎜🎜 수백만 개의 웹 페이지를 크롤링하는 크롤러 서비스를 개발한다고 가정해 보겠습니다. 특정 웹 페이지가 크롤링되었는지 확인하는 방법 데이터베이스와 HashMap을 사용하는 것 외에도 Bloom 필터를 사용하여 이를 수행할 수 있습니다. . 다른 방법에 비해 Bloom 필터는 공간을 거의 차지하지 않으며 삽입 및 쿼리 시간이 매우 빠릅니다. 🎜🎜Bloom 필터는 요소가 집합에 있는지 확인하는 데 사용됩니다. BitSet🎜🎜🎜🎜을 사용하여 데이터를 삽입하면 값이 여러 번 해시되고 BitSet의 해당 위치는 1🎜🎜🎜🎜쿼리할 때 값은 여러 번 해시되어 모두 비교됩니다. 비트가 1인지 여부, 그렇다면 존재합니다. 🎜🎜🎜🎜 블룸 필터는 특정 오판율이 있어 정확한 쿼리 시나리오에는 적합하지 않습니다. 또한 요소 삭제는 지원되지 않습니다. 일반적으로 URL 중복 제거, 스팸 필터링, 캐시 분석 방지와 같은 시나리오에 사용됩니다. 🎜🎜Redis에서는 내장된 BitSet 구현을 사용할 수 있으며 Lua 스크립트의 원자성을 사용하여 여러 쿼리 데이터 불일치를 방지할 수 있습니다. 🎜
func main() { rdb := redis.NewClient(&redis.Options{ Addr: "localhost:6379", Password: "123456", DB: 0, // use default DB }) r, err := NewRedisLimiter(rdb, 1, 2, "testrate") if err != nil { log.Fatal(err) } r.Reset() for i := 0; i < 5; i++ { err := r.Wait(context.TODO()) log.Printf("worker %d allowed: %v", i, err) } } // output // 2022/07/22 12:50:31 worker 0 allowed: <nil> // 2022/07/22 12:50:31 worker 1 allowed: <nil> // 2022/07/22 12:50:32 worker 2 allowed: <nil> // 2022/07/22 12:50:33 worker 3 allowed: <nil> // 2022/07/22 12:50:34 worker 4 allowed: <nil>
테스트 실행
rrreee🎜분산 속도 제한기🎜🎜는 golang.org/x/time/rate 패키지에 토큰 버킷 기반 속도 제한기를 제공합니다. 분산 환경에서 전류 제한을 구현하려면 Redis Lua 스크립트를 기반으로 구현할 수 있습니다. 🎜🎜토큰 버킷의 주요 원리는 다음과 같습니다. 🎜🎜🎜🎜 토큰 버킷의 용량이 폭발하고 매초마다 qps 비율로 토큰이 배치된다고 가정합니다.🎜🎜🎜🎜토큰은 처음에 채워집니다. 토큰으로, 토큰이 오버플로되면 바로 채워집니다. 토큰을 요청할 때 버킷에 충분한 토큰이 있으면 허용하고, 그렇지 않으면 거부됩니다.🎜🎜🎜🎜burst==qps일 때 흐름 제한은 qps를 엄격히 준수합니다. 버스트>qps인 경우 특정 트래픽 버스트가 허용될 수 있습니다🎜

这里主要参考了官方rate包的实现，将核心逻辑改为Lua实现。

--- 相关Key
--- limit rate key值，对应value为当前令牌数
local limit_key = KEYS[1]
--- 输入参数
--[[
qps: 每秒请求数;
burst: 令牌桶容量;
now: 当前Timestamp;
cost: 请求令牌数;
max_wait: 最大等待时间
--]]
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
--- 上次修改时间
local last_time = redis.call("hget", limit_key, "last_time")
if not last_time then
	last_time = 0
end
--- 最新等待时间
local last_event = redis.call("hget", limit_key, "last_event")
if not last_event then
	last_event = 0
end
--- 通过当前时间与上次修改时间的差值，qps计算出当前时间得令牌数
local delta = math.max(0, now-last_time)
local new_tokens = math.min(burst, delta * qps + tokens)
new_tokens = new_tokens - cost --- 最新令牌数，减少请求令牌
--- 如果最新令牌数小于0，计算需要等待的时间
local wait_period = 0
if new_tokens < 0 and qps > 0 then
	wait_period = wait_period - new_tokens / qps
end
wait_period = math.ceil(wait_period)
local time_act = now + wait_period --- 满足等待间隔的时间戳
--- 允许请求有两种情况
--- 当请求令牌数小于burst, 等待时间不超过最大等待时间，可以通过补充令牌满足请求
--- qps为0时，只要最新令牌数不小于0即可
local ok = (cost <= burst and wait_period <= max_wait and qps > 0) or (qps == 0 and new_tokens >= 0)
--- 设置对应值
if ok then
	redis.call("set", limit_key, new_tokens)
	redis.call("set", last_time_key, now)
	redis.call("set", last_event_key, time_act)
end
--- 返回列表，{是否允许， 等待时间}
return {ok, wait_period}

在Golang中的相关接口Allow、AllowN、Wait等都是通过调用reserveN实现

// 调用lua脚本
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
	}
	//...
	return &Reservation{
		ok:        allow == 1,
		lim:       lim,
		tokens:    n,
		timeToAct: now.Add(time.Duration(wait) * time.Second),
	}, nil
}

运行测试

func main() {
	rdb := redis.NewClient(&redis.Options{
		Addr:     "localhost:6379",
		Password: "123456",
		DB:       0, // use default DB
	})
	r, err := NewRedisLimiter(rdb, 1, 2, "testrate")
	if err != nil {
		log.Fatal(err)
	}
	r.Reset()
	for i := 0; i < 5; i++ {
		err := r.Wait(context.TODO())
		log.Printf("worker %d allowed: %v", i, err)
	}
}
// output
// 2022/07/22 12:50:31 worker 0 allowed: <nil>
// 2022/07/22 12:50:31 worker 1 allowed: <nil>
// 2022/07/22 12:50:32 worker 2 allowed: <nil>
// 2022/07/22 12:50:33 worker 3 allowed: <nil>
// 2022/07/22 12:50:34 worker 4 allowed: <nil>

前两个请求在burst内，直接可以获得，后面的请求按照qps的速率生成。

其他

Redis还可用于全局计数、去重以及发布订阅等不同情境。参考Redis官方提供的模块，可以通过加载这些模块实现过滤、限流等特性。

위 내용은 Golang 분산 애플리케이션에서 Redis를 사용하는 방법의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!

성명：

이 기사는 yisu.com에서 복제됩니다. 침해가 있는 경우 admin@php.cn으로 문의하시기 바랍니다. 삭제

이전 기사：Docker PHP 컨테이너에 Redis 확장을 설치하는 방법다음 기사：Docker PHP 컨테이너에 Redis 확장을 설치하는 방법

Golang 분산 애플리케이션에서 Redis를 사용하는 방법

Text

분산 잠금

运行测试

分布式过滤器

运行测试

分布式限流器

테스트 실행

테스트 실행

运行测试

其他

관련 기사