원본 기사는 VictoriaMetrics 블로그에 게시되어 있습니다: https://victoriametrics.com/blog/go-singleflight/
이 게시물은 Go의 동시성 처리에 관한 시리즈의 일부입니다.
따라서 동일한 데이터를 요청하는 여러 요청이 동시에 들어오는 경우 기본 동작은 각 요청이 동일한 정보를 얻기 위해 개별적으로 데이터베이스로 이동하는 것입니다. . 이것이 의미하는 바는 결국 동일한 쿼리를 여러 번 실행하게 된다는 것인데, 솔직히 말해서 이는 비효율적입니다.
결국 데이터베이스에 불필요한 로드가 발생하여 모든 것이 느려질 수 있지만 이를 해결할 수 있는 방법이 있습니다.
첫 번째 요청만 실제로 데이터베이스로 이동한다는 개념입니다. 나머지 요청은 첫 번째 요청이 완료될 때까지 기다립니다. 초기 요청에서 데이터가 반환되면 다른 요청도 동일한 결과를 얻습니다. 추가 쿼리가 필요하지 않습니다.
이제 이 게시물이 무엇에 관한 것인지 꽤 잘 아셨죠?
Go의 Singleflight 패키지는 방금 이야기한 내용을 정확하게 처리하기 위해 특별히 제작되었습니다. 그리고 미리 말씀드리지만, 표준 라이브러리의 일부는 아니지만 Go 팀에서 유지 관리하고 개발합니다.
singleflight가 하는 일은 데이터베이스에서 데이터를 가져오는 것과 같은 고루틴 중 하나만 실제로 작업을 실행하도록 하는 것입니다. 특정 순간에 동일한 데이터 조각("키"라고 함)에 대해 단 한 번의 "진행 중"(진행 중인) 작업만 허용합니다.
따라서 해당 작업이 계속 진행되는 동안 다른 고루틴이 동일한 데이터(동일한 키)를 요청하면 기다리게 됩니다. 그런 다음 첫 번째 작업이 완료되면 작업을 다시 실행할 필요 없이 다른 모든 작업도 동일한 결과를 얻습니다.
자, 이야기는 이쯤 하고, 싱글플라이트가 실제로 어떻게 작동하는지 알아보기 위해 간단한 데모를 살펴보겠습니다.
var callCount atomic.Int32
var wg sync.WaitGroup
// Simulate a function that fetches data from a database
func fetchData() (interface{}, error) {
callCount.Add(1)
time.Sleep(100 * time.Millisecond)
return rand.Intn(100), nil
}
// Wrap the fetchData function with singleflight
func fetchDataWrapper(g *singleflight.Group, id int) error {
defer wg.Done()
time.Sleep(time.Duration(id) * 40 * time.Millisecond)
v, err, shared := g.Do("key-fetch-data", fetchData)
if err != nil {
return err
}
fmt.Printf("Goroutine %d: result: %v, shared: %v\n", id, v, shared)
return nil
}
func main() {
var g singleflight.Group
// 5 goroutines to fetch the same data
const numGoroutines = 5
wg.Add(numGoroutines)
for i := 0; i < numGoroutines; i++ {
go fetchDataWrapper(&g, i)
}
wg.Wait()
fmt.Printf("Function was called %d times\n", callCount.Load())
}
// Output:
// Goroutine 0: result: 90, shared: true
// Goroutine 2: result: 90, shared: true
// Goroutine 1: result: 90, shared: true
// Goroutine 3: result: 13, shared: true
// Goroutine 4: result: 13, shared: true
// Function was called 2 times
현재 상황:
우리는 5개의 고루틴이 60ms 간격으로 거의 동시에 동일한 데이터를 가져오려고 시도하는 상황을 시뮬레이션하고 있습니다. 단순함을 유지하기 위해 난수를 사용하여 데이터베이스에서 가져온 데이터를 모방합니다.
singleflight.Group을 사용하면 첫 번째 고루틴만 실제로 fetchData()를 실행하고 나머지는 결과를 기다립니다.
v, err, shared := g.Do("key-fetch-data", fetchData) 줄은 이러한 요청을 추적하기 위해 고유 키("key-fetch-data")를 할당합니다. 따라서 첫 번째 고루틴이 여전히 데이터를 가져오는 동안 다른 고루틴이 동일한 키를 요청하면 새 호출을 시작하는 대신 결과를 기다립니다.
첫 번째 호출이 끝나면 출력에서 볼 수 있듯이 대기 중인 모든 고루틴은 동일한 결과를 얻습니다. 데이터를 요청하는 고루틴이 5개 있었지만 fetchData는 두 번만 실행되었는데, 이는 엄청난 성능 향상입니다.
공유 플래그는 결과가 여러 고루틴에서 재사용되었음을 확인합니다.
"근데 왜 첫 번째 고루틴에서는 공유 플래그가 true인가요? 대기 중인 사람만 공유 == true일 거라고 생각했어요?"
예, 대기 중인 고루틴만 == true를 공유해야 한다고 생각한다면 이는 다소 직관에 어긋난다고 느껴질 수 있습니다.
중요한 점은 g.Do의 공유 변수가 결과가 여러 호출자 간에 공유되었는지 여부를 알려준다는 것입니다. 기본적으로 "이 결과는 두 명 이상의 발신자가 사용했습니다."라는 뜻입니다. 누가 함수를 실행했는지가 아니라 결과가 여러 고루틴에서 재사용되었다는 신호일 뿐입니다.
"캐시가 있는데 왜 싱글플라이트가 필요한가요?"
짧은 대답은 다음과 같습니다. 캐시와 단일 비행은 서로 다른 문제를 해결하며 실제로 서로 잘 작동합니다.
외부 캐시(예: Redis 또는 Memcached)를 사용하는 설정에서 Singleflight는 데이터베이스뿐만 아니라 캐시 자체에도 추가 보호 계층을 추가합니다.
또한 Singleflight는 캐시 미스 폭풍("캐시 스탬피드"라고도 함)으로부터 보호하는 데 도움이 됩니다.
일반적으로 요청에서 데이터를 요청할 때 데이터가 캐시에 있으면 좋습니다. 이는 캐시 적중입니다. 데이터가 캐시에 없으면 캐시 누락입니다. 캐시가 재구축되기 전에 10,000개의 요청이 동시에 시스템에 도달한다고 가정하면 데이터베이스가 동시에 10,000개의 동일한 쿼리로 갑자기 중단될 수 있습니다.
이 피크 기간 동안 Singleflight는 10,000개의 요청 중 하나만 실제로 데이터베이스에 도달하도록 보장합니다.
그러나 나중에 내부 구현 섹션에서는 단일 비행이 모든 고루틴에 대한 단일 경합 지점이 될 수 있는 비행 중 호출의 지도를 보호하기 위해 전역 잠금을 사용하는 것을 볼 수 있습니다. 특히 높은 동시성을 처리하는 경우 작업 속도가 느려질 수 있습니다.
아래 모델은 여러 CPU가 있는 시스템에 더 잘 작동할 수 있습니다.
이 설정에서는 캐시 누락이 발생한 경우에만 단일 비행을 사용합니다.
singleflight를 사용하려면 먼저 특정 키에 연결된 지속적인 함수 호출을 추적하는 핵심 구조인 그룹 개체를 만듭니다.
중복 통화를 방지하는 데 도움이 되는 두 가지 주요 방법이 있습니다.
우리는 이미 데모에서 g.Do()를 사용하는 방법을 살펴보았습니다. 수정된 래퍼 함수로 g.DoChan()을 사용하는 방법을 확인해 보겠습니다.
var callCount atomic.Int32
var wg sync.WaitGroup
// Simulate a function that fetches data from a database
func fetchData() (interface{}, error) {
callCount.Add(1)
time.Sleep(100 * time.Millisecond)
return rand.Intn(100), nil
}
// Wrap the fetchData function with singleflight
func fetchDataWrapper(g *singleflight.Group, id int) error {
defer wg.Done()
time.Sleep(time.Duration(id) * 40 * time.Millisecond)
v, err, shared := g.Do("key-fetch-data", fetchData)
if err != nil {
return err
}
fmt.Printf("Goroutine %d: result: %v, shared: %v\n", id, v, shared)
return nil
}
func main() {
var g singleflight.Group
// 5 goroutines to fetch the same data
const numGoroutines = 5
wg.Add(numGoroutines)
for i := 0; i < numGoroutines; i++ {
go fetchDataWrapper(&g, i)
}
wg.Wait()
fmt.Printf("Function was called %d times\n", callCount.Load())
}
// Output:
// Goroutine 0: result: 90, shared: true
// Goroutine 2: result: 90, shared: true
// Goroutine 1: result: 90, shared: true
// Goroutine 3: result: 13, shared: true
// Goroutine 4: result: 13, shared: true
// Function was called 2 times
// Wrap the fetchData function with singleflight using DoChan
func fetchDataWrapper(g *singleflight.Group, id int) error {
defer wg.Done()
ch := g.DoChan("key-fetch-data", fetchData)
res := <-ch
if res.Err != nil {
return res.Err
}
fmt.Printf("Goroutine %d: result: %v, shared: %v\n", id, res.Val, res.Shared)
return nil
}
솔직히 여기서 DoChan()을 사용하는 것은 Do()에 비해 크게 변하지 않습니다. 기본적으로 동일한 채널을 차단하는 채널 수신 작업(<-ch)으로 결과를 기다리고 있기 때문입니다. 방법입니다.
DoChan()이 빛을 발하는 곳은 고루틴을 차단하지 않고 작업을 시작하고 다른 작업을 수행하려는 경우입니다. 예를 들어 채널을 사용하면 시간 초과나 취소를 더 깔끔하게 처리할 수 있습니다.
package singleflight
type Result struct {
Val interface{}
Err error
Shared bool
}
이 예에서는 실제 시나리오에서 발생할 수 있는 몇 가지 문제도 제기합니다.
예, Singleflight는 진행 중인 실행을 취소할 수 있는 group.Forget(key) 메서드를 사용하여 이러한 상황을 처리하는 방법을 제공합니다.
Forget() 메서드는 진행 중인 함수 호출을 추적하는 내부 맵에서 키를 제거합니다. 이는 키를 "무효화"하는 것과 비슷하므로 해당 키로 g.Do()를 다시 호출하면 이전 실행이 완료될 때까지 기다리는 대신 마치 새로운 요청인 것처럼 함수가 실행됩니다.
Forget()을 사용하도록 예제를 업데이트하고 함수가 실제로 몇 번 호출되는지 살펴보겠습니다.
var callCount atomic.Int32
var wg sync.WaitGroup
// Simulate a function that fetches data from a database
func fetchData() (interface{}, error) {
callCount.Add(1)
time.Sleep(100 * time.Millisecond)
return rand.Intn(100), nil
}
// Wrap the fetchData function with singleflight
func fetchDataWrapper(g *singleflight.Group, id int) error {
defer wg.Done()
time.Sleep(time.Duration(id) * 40 * time.Millisecond)
v, err, shared := g.Do("key-fetch-data", fetchData)
if err != nil {
return err
}
fmt.Printf("Goroutine %d: result: %v, shared: %v\n", id, v, shared)
return nil
}
func main() {
var g singleflight.Group
// 5 goroutines to fetch the same data
const numGoroutines = 5
wg.Add(numGoroutines)
for i := 0; i < numGoroutines; i++ {
go fetchDataWrapper(&g, i)
}
wg.Wait()
fmt.Printf("Function was called %d times\n", callCount.Load())
}
// Output:
// Goroutine 0: result: 90, shared: true
// Goroutine 2: result: 90, shared: true
// Goroutine 1: result: 90, shared: true
// Goroutine 3: result: 13, shared: true
// Goroutine 4: result: 13, shared: true
// Function was called 2 times
Goroutine 0과 Goroutine 1은 모두 동일한 키("key-fetch-data")를 사용하여 Do()를 호출하며 해당 요청은 하나의 실행으로 결합되고 결과는 두 goroutine 간에 공유됩니다.
반면에 Goroutine 2는 Do()를 실행하기 전에 Forget()을 호출합니다. 이렇게 하면 "key-fetch-data"와 관련된 이전 결과가 모두 지워지고 함수의 새로운 실행이 시작됩니다.
요약하자면, 싱글 플라이트(singleflight)는 유용하지만 여전히 다음과 같은 일부 예외적인 경우가 있을 수 있습니다.
우리가 논의한 모든 문제를 발견했다면 다음 섹션으로 넘어가서 실제로 단일 비행이 내부적으로 어떻게 작동하는지 논의해 보겠습니다.
singleflight를 사용하여 내부적으로 어떻게 작동하는지에 대한 기본 아이디어를 이미 알고 있을 수 있으며, Singleflight의 전체 구현은 약 150줄의 코드에 불과합니다.
기본적으로 모든 고유 키는 실행을 관리하는 구조체를 갖습니다. 고루틴이 Do()를 호출하고 키가 이미 존재한다는 것을 발견하면 첫 번째 실행이 완료될 때까지 해당 호출이 차단되며 구조는 다음과 같습니다.
// Wrap the fetchData function with singleflight using DoChan
func fetchDataWrapper(g *singleflight.Group, id int) error {
defer wg.Done()
ch := g.DoChan("key-fetch-data", fetchData)
res := <-ch
if res.Err != nil {
return res.Err
}
fmt.Printf("Goroutine %d: result: %v, shared: %v\n", id, res.Val, res.Shared)
return nil
}
여기에서는 두 가지 동기화 기본 요소가 사용됩니다.
여기에서는 group.Do() 메서드에 중점을 두겠습니다. 다른 메서드인 group.DoChan()도 비슷한 방식으로 작동하기 때문입니다. group.Forget() 메소드도 맵에서 키를 제거하기 때문에 간단합니다.
group.Do()를 호출하면 가장 먼저 하는 일은 전체 호출 맵(g.mu)을 잠그는 것입니다.
"그렇게 성능면에서 나쁘지 않나?"
예, 단일 비행이 전체 키를 잠그기 때문에 모든 경우에 성능에 이상적이지는 않을 수 있습니다(항상 먼저 벤치마킹하는 것이 좋습니다). 더 나은 성능을 목표로 하거나 대규모로 작업하는 경우 키를 분할하거나 배포하는 것이 좋은 접근 방식입니다. 하나의 단일 비행 그룹을 사용하는 대신 "다중 비행"을 수행하는 것과 같이 여러 그룹에 부하를 분산할 수 있습니다
참고로 shardedsingleflight 저장소를 확인하세요.
이제 잠금이 설정되면 해당 키에 대해 이미 진행 중이거나 완료된 호출이 있는 경우 그룹은 내부 지도(g.m)를 확인합니다. 이 지도는 해당 작업에 매핑된 키를 사용하여 진행 중이거나 완료된 작업을 추적합니다.
키가 발견되면(다른 고루틴이 이미 작업을 실행 중임) 새 호출을 시작하는 대신 단순히 카운터(c.dups)를 증가시켜 중복 요청을 추적합니다. 그런 다음 고루틴은 잠금을 해제하고 연결된 WaitGroup에서 call.wg.Wait()를 호출하여 원래 작업이 완료될 때까지 기다립니다.
원래 작업이 완료되면 이 고루틴은 결과를 가져오고 작업 다시 실행을 방지합니다.
var callCount atomic.Int32
var wg sync.WaitGroup
// Simulate a function that fetches data from a database
func fetchData() (interface{}, error) {
callCount.Add(1)
time.Sleep(100 * time.Millisecond)
return rand.Intn(100), nil
}
// Wrap the fetchData function with singleflight
func fetchDataWrapper(g *singleflight.Group, id int) error {
defer wg.Done()
time.Sleep(time.Duration(id) * 40 * time.Millisecond)
v, err, shared := g.Do("key-fetch-data", fetchData)
if err != nil {
return err
}
fmt.Printf("Goroutine %d: result: %v, shared: %v\n", id, v, shared)
return nil
}
func main() {
var g singleflight.Group
// 5 goroutines to fetch the same data
const numGoroutines = 5
wg.Add(numGoroutines)
for i := 0; i < numGoroutines; i++ {
go fetchDataWrapper(&g, i)
}
wg.Wait()
fmt.Printf("Function was called %d times\n", callCount.Load())
}
// Output:
// Goroutine 0: result: 90, shared: true
// Goroutine 2: result: 90, shared: true
// Goroutine 1: result: 90, shared: true
// Goroutine 3: result: 13, shared: true
// Goroutine 4: result: 13, shared: true
// Function was called 2 times
해당 키에 대해 다른 고루틴이 작동하지 않는 경우 현재 고루틴이 작업 실행을 담당합니다.
이 시점에서는 새 호출 개체를 생성하여 맵에 추가하고 해당 WaitGroup을 초기화합니다. 그런 다음 뮤텍스를 잠금 해제하고 도우미 메서드 g.doCall(c, key, fn)을 통해 직접 작업 실행을 진행합니다. 작업이 완료되면 wg.Wait() 호출에 의해 대기 중인 모든 고루틴이 차단 해제됩니다.
오류를 처리하는 방법을 제외하면 다음과 같은 세 가지 시나리오가 있습니다.
여기서 도우미 메서드 g.doCall()이 좀 더 영리해지기 시작합니다.
"잠깐, 런타임.Goexit()이 뭐죠?"
코드를 살펴보기 전에 빠르게 설명하자면, Runtime.Goexit()은 고루틴 실행을 중지하는 데 사용됩니다.
고루틴이 Goexit()을 호출하면 중지되고 지연된 모든 함수는 평소와 마찬가지로 후입선출(LIFO) 순서로 계속 실행됩니다. 패닉과 유사하지만 몇 가지 차이점이 있습니다.
여기에 흥미로운 특징이 있습니다(우리 주제와 직접적으로 관련은 없지만 언급할 가치가 있음). 메인 고루틴(main() 내부와 같은)에서 Runtime.Goexit()을 호출하는 경우 다음을 확인하세요.
var callCount atomic.Int32
var wg sync.WaitGroup
// Simulate a function that fetches data from a database
func fetchData() (interface{}, error) {
callCount.Add(1)
time.Sleep(100 * time.Millisecond)
return rand.Intn(100), nil
}
// Wrap the fetchData function with singleflight
func fetchDataWrapper(g *singleflight.Group, id int) error {
defer wg.Done()
time.Sleep(time.Duration(id) * 40 * time.Millisecond)
v, err, shared := g.Do("key-fetch-data", fetchData)
if err != nil {
return err
}
fmt.Printf("Goroutine %d: result: %v, shared: %v\n", id, v, shared)
return nil
}
func main() {
var g singleflight.Group
// 5 goroutines to fetch the same data
const numGoroutines = 5
wg.Add(numGoroutines)
for i := 0; i < numGoroutines; i++ {
go fetchDataWrapper(&g, i)
}
wg.Wait()
fmt.Printf("Function was called %d times\n", callCount.Load())
}
// Output:
// Goroutine 0: result: 90, shared: true
// Goroutine 2: result: 90, shared: true
// Goroutine 1: result: 90, shared: true
// Goroutine 3: result: 13, shared: true
// Goroutine 4: result: 13, shared: true
// Function was called 2 times
Goexit()가 기본 고루틴을 종료하지만 다른 고루틴이 아직 실행 중인 경우 적어도 하나의 고루틴이 활성화되어 있는 한 Go 런타임이 활성 상태로 유지되므로 프로그램은 계속 진행됩니다. 그러나 고루틴이 하나도 남지 않으면 "고루틴 없음" 오류가 발생하며 충돌이 발생합니다. 이는 일종의 재미있는 코너 케이스입니다.
이제 코드로 돌아가서, Runtime.Goexit()가 현재 고루틴만 종료하고 Recover()가 이를 포착할 수 없는 경우, 호출되었는지 어떻게 감지합니까?
핵심은 Runtime.Goexit()가 호출될 때 그 이후의 모든 코드가 실행되지 않는다는 사실에 있습니다.
// Wrap the fetchData function with singleflight using DoChan
func fetchDataWrapper(g *singleflight.Group, id int) error {
defer wg.Done()
ch := g.DoChan("key-fetch-data", fetchData)
res := <-ch
if res.Err != nil {
return res.Err
}
fmt.Printf("Goroutine %d: result: %v, shared: %v\n", id, res.Val, res.Shared)
return nil
}
위의 경우, running.Goexit()를 호출한 후 NormalReturn = true 줄이 실행되지 않습니다. 따라서 defer 내부에서 NormalReturn이 여전히 false인지 확인하여 특수 메소드가 호출되었음을 감지할 수 있습니다.
다음 단계는 작업이 패닉 상태인지 여부를 파악하는 것입니다. 이를 위해 일반 반환으로 복구()를 사용하지만, Singleflight의 실제 코드는 좀 더 미묘합니다.
package singleflight
type Result struct {
Val interface{}
Err error
Shared bool
}
recover 블록 내에서 Recover = true를 직접 설정하는 대신, Recover() 블록 이후에 Recovered를 마지막 줄로 설정하면 이 코드가 약간 멋져집니다.
그렇다면 이것이 왜 작동하는 걸까요?
runtime.Goexit()이 호출되면 패닉()과 마찬가지로 전체 고루틴이 종료됩니다. 그러나panic()이 복구되면 전체 고루틴이 아닌nic()과 Recover() 사이의 함수 체인만 종료됩니다.
이것이 복구 = true가 복구()를 포함하는 defer 외부에서 설정되는 이유입니다. 두 가지 경우에만 실행됩니다: 함수가 정상적으로 완료되거나 패닉이 복구될 때, 런타임.Goexit()가 호출될 때는 실행되지 않습니다.
앞으로 각 사례를 어떻게 처리하는지 논의해보겠습니다.
func fetchDataWrapperWithTimeout(g *singleflight.Group, id int) error {
defer wg.Done()
ch := g.DoChan("key-fetch-data", fetchData)
select {
case res := <-ch:
if res.Err != nil {
return res.Err
}
fmt.Printf("Goroutine %d: result: %v, shared: %v\n", id, res.Val, res.Shared)
case <-time.After(50 * time.Millisecond):
return fmt.Errorf("timeout waiting for result")
}
return nil
}
실행 중에 작업 패닉이 발생하면 패닉이 포착되어 c.err에 패닉 값과 스택 추적을 모두 보유하는 패닉 오류로 저장됩니다. Singleflight는 패닉을 포착하여 우아하게 정리하지만 이를 삼키지는 않고 해당 상태를 처리한 후 패닉을 다시 발생시킵니다.
즉, 작업을 실행하는 고루틴(작업을 시작한 첫 번째 고루틴)에서 패닉이 발생하고 결과를 기다리는 다른 모든 고루틴도 패닉 상태가 된다는 의미입니다.
이러한 패닉은 개발자의 코드에서 발생하므로 적절하게 처리하는 것은 우리의 몫입니다.
이제 고려해야 할 특별한 경우가 있습니다. 다른 고루틴이 group.DoChan() 메서드를 사용하고 채널을 통해 결과를 기다리는 경우입니다. 이 경우, 단일 비행은 해당 고루틴에서 당황할 수 없습니다. 대신 복구할 수 없는 패닉(go 패닉(e))을 수행하여 애플리케이션이 중단됩니다.
마지막으로, Runtime.Goexit()이라는 작업이 있는 경우 고루틴이 이미 종료되는 중이므로 추가 조치를 취할 필요가 없으며 방해하지 않고 종료되도록 놔두기만 하면 됩니다.
거의 전부입니다. 우리가 논의한 특별한 경우를 제외하면 그다지 복잡한 것은 없습니다.
안녕하세요, 저는 VictoriaMetrics의 소프트웨어 엔지니어인 Phuong Le입니다. 위의 글쓰기 스타일은 명확성과 단순성에 중점을 두고 학문적 정확성에 항상 완벽하게 부합하지는 않더라도 이해하기 쉬운 방식으로 개념을 설명합니다.
오래된 내용을 발견했거나 궁금한 점이 있으면 주저하지 말고 문의하세요. X(@func25)로 DM을 보내주세요.
당신이 관심을 가질 만한 다른 게시물:
서비스를 모니터링하고 지표를 추적하고 모든 것이 어떻게 수행되는지 확인하려면 VictoriaMetrics를 확인해 보세요. 이는 인프라를 감시할 수 있는 빠르고 오픈 소스이며 비용을 절약하는 방법입니다.
저희는 Go와 Go 생태계에 대한 연구, 실험, 지식 공유를 좋아하는 Gophers입니다.
위 내용은 Go Singleflight는 DB가 아닌 코드에서 용해됩니다.의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!
![[웹 프런트엔드] Node.js 빠른 시작](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
