[golang] go context, go-cache를 활용한 캐시 서버 만들기

이번게시글에서는 go-context와 go-cache를 활용한 간단한 캐시서버를 구축하는 예제입니다.

소스코드는 github.com/jaeho310/proxy-tut 에서 확인할 수 있습니다.

GitHub - jaeho310/proxy-tut

 

1. 캐시 서버의 기능

(1). 클라이언트에게 요청이 오면 캐시되었는지 확인한후, 캐시되어있다면 본서버로 요청을 보내지 않고 데이터를 내려줍니다. 

(2). 동일한 요청이 여러개 들어올 시(본서버에서 응답을 받기전이라 캐시되지 않은경우) 동일한 요청은 본서버에 한번만 보내고 나머지 요청은 블락시키고 응답을 받으면 공유하여 내려줍니다.

 

request는 고루틴이 실행하므로 해당 루틴을 channel을 사용하여 블락시키고 동일한 response를 내려줘서 1번을 구현합니다.

하지만 채널과 고루틴은 1:1 관계이므로 go context를 사용하여 같은 context에 포함된 goroutine을 한번에 제어합니다.

2. 캐시 저장소 구성 및 저장규칙

(1) CacheRepository

go cache를 사용하여 관리하여 본서버에서 데이터를 받으면 1분간 데이터를 캐시합니다.

캐시를 저장하느라 클라이언트에게 데이터를 늦게 주면 안되므로 저장로직은 고루틴(스레드)으로 돌아갑니다.

스케일 아웃을 고려한다면 구현체로 redis등의 global 저장소를 사용합니다.

 

(2) TemporaryRepository

동일한 요청이 본서버의 응답을 받았다면 데이터가 캐시되어있겠지만, 본서버에서 응답이 오기전(캐시하기전) 동일한 요청이 또 들어왔을때를 처리하기 위함입니다. (본서버에서 응답을 내려주는데 1초가 걸린다면 1초안에 온 100개의 동일한 요청은 본서버에 들어갈 필요 x) 본서버에 들어간 요청과 동일한 요청은 block시킨후 하나의 응답이 오면 모두 response를 내려줍니다.

본서버에서 응답이 오면 고루틴(스레드)를 사용하여 데이터를 캐시하므로 고루틴이 돌기전 들어온 동일한 요청의 데이터를 캐시(2초간)하는 역할도 포함됩니다.

 

3.  output

프록시서버: [hello] 요청을 본서버에 보냈습니다. 2초동안 동일한 요청은 본서버에 보내지 않습니다 
프록시서버: [hello] 요청은 이미 본서버에 보낸 요청입니다. 응답이 오면 동일한 결과를 내립니다. 
프록시서버: [hello] 요청은 이미 본서버에 보낸 요청입니다. 응답이 오면 동일한 결과를 내립니다. 
본서버: [hello] 요청을 받았습니다 결과값을 내려줍니다. 
클라이언트: 결과 데이터: world
클라이언트: 결과 데이터: world
클라이언트: 결과 데이터: world
프록시서버: 본서버에서 response가 왔으므로 {hello: world}의 데이터를 캐시합니다
프록시서버: [hello]는 캐시되어있으므로 [world]를 내려줍니다
클라이언트: 결과 데이터: world
프록시서버: [hello]는 캐시되어있으므로 [world]를 내려줍니다
클라이언트: 결과 데이터: world

 

4.  패키지구조

.
|-- go.mod
|-- go.sum
|-- gateway
|   |-- data.gateway.go
|   `-- data.gateway.impl.go
|-- main.go
|-- model
|   `-- response.ctx.model.go
|-- repository
|   |-- cache.repository.go
|   |-- cache.repository.impl.go
|   |-- temporary.repository.go
|   `-- temporary.repository.impl.go
`-- service
    |-- data.service.go
    |-- data.service.impl.go
    `-- proxy.data.service.impl.go

 

5. ProxyDataServiceImpl

package service

import (
	"fmt"
	"proxy-tut/gateway"
	"proxy-tut/repository"
)

type ProxyDataServiceImpl struct {
	dataGateway         gateway.DataGateway
	temporaryRepository repository.TemporaryRepository
	cacheRepository     repository.CacheRepository
}

func (proxyServiceImpl ProxyDataServiceImpl) New(dataGateway gateway.DataGateway) *ProxyDataServiceImpl {
	temporaryRepositoryImpl := repository.TemporaryRepositoryImpl{}.New()
	cacheRepositoryImpl := repository.CacheRepositoryImpl{}.New()
	return &ProxyDataServiceImpl{dataGateway, temporaryRepositoryImpl, cacheRepositoryImpl}
}

func (proxyServiceImpl *ProxyDataServiceImpl) GetDataWithParam(param string) (string, error) {
	var responseData string
	var err error
	// 캐시된 요청이면 바로 응답을 내려준다.
	responseData = proxyServiceImpl.cacheRepository.GetData(param)
	if len(responseData) != 0 {
		fmt.Printf("프록시서버: [%s]는 캐시되어있으므로 [%s]를 내려줍니다\n", param, responseData)
		return responseData, nil
	}
	// 캐시되지 않은 요청이면 최초의 요청인지 확인
	first, responseModelWithCtx := proxyServiceImpl.temporaryRepository.GetResponseModel(param)
	defer responseModelWithCtx.Cancel()
	if first {
		// 최초의 요청인 경우
		fmt.Printf("프록시서버: [%s] 요청을 본서버에 보냈습니다. " +
			"2초동안 동일한 요청은 본서버에 보내지 않습니다 \n", param)
		responseData, err = proxyServiceImpl.dataGateway.GetDataFromDataServer(param)
		if err != nil {
			return "", err
		}
		responseModelWithCtx.Value = responseData
		// 최초의 요청이 본서버에서 데이터를 받으면 나머지 요청들과 데이터를 공유해 내려준다.
		responseModelWithCtx.Cancel()

		// 클라이언트에게 응답은 바로 가야하므로 캐시하는 로직은 고루틴으로 처리
		go func() {
			proxyServiceImpl.cacheRepository.SetData(param, responseData)
		}()
	} else {
		fmt.Printf("프록시서버: [%s] 요청은 이미 본서버에 보낸 요청입니다." +
			" 응답이 오면 동일한 결과를 내립니다. \n", param)
		// 최초의 요청이 아니라면 2초동안은 본서버에 request를 보내지않고 block
		<-responseModelWithCtx.Ctx.Done()
		responseData = responseModelWithCtx.Value
	}

	return responseData, nil
}

캐시된 요청이라면 바로 응답을 내려주며

캐시되지 않은 요청은 최초의 요청인지 확인, 최초의 요청이라면 param당 하나의 go context를 발급하여 최초의 요청이 아닌 고루틴을 블락시킵니다.

최초의 요청이 본서버에서 데이터를 받아오면 문맥을 종료시켜 중복된 요청의 블락을 풀어주고, 데이터를 공유하여 response를 내려줍니다.

response를 내려줄때 고루틴을 하나 생성하여 cacheRepository에 param: data 를 캐시해놓습니다. 

 

6. TemporaryRepositryImpl

package repository

import (
	"context"
	"github.com/patrickmn/go-cache"
	"proxy-tut/model"
	"sync"
	"time"
)

type TemporaryRepositoryImpl struct {
	cache *cache.Cache
	mu    *sync.Mutex
}

func (TemporaryRepositoryImpl) New() *TemporaryRepositoryImpl {
	c := cache.New(2*time.Second, 1*time.Second)
	return &TemporaryRepositoryImpl{c, &sync.Mutex{}}
}

func (temporaryRepositoryImpl *TemporaryRepositoryImpl) setData(key string, value *model.ResponseCtxModel) {
	temporaryRepositoryImpl.cache.Set(key, value, cache.DefaultExpiration)
}

func (temporaryRepositoryImpl *TemporaryRepositoryImpl) GetResponseModel(key string) (bool, *model.ResponseCtxModel) {
	temporaryRepositoryImpl.mu.Lock()
	defer temporaryRepositoryImpl.mu.Unlock()
	data, found := temporaryRepositoryImpl.cache.Get(key)
	if found {
		return false, data.(*model.ResponseCtxModel)
	} else {
		ctx, cancel := context.WithCancel(context.Background())
		temporaryModel := &model.ResponseCtxModel{Value: "", Cancel: cancel, Ctx: ctx}
		temporaryRepositoryImpl.setData(key, temporaryModel)
		return true, temporaryModel
	}

}

임시저장소는 2초간 중복된 데이터를 본 서버에 보내지 않도록 해줍니다.

param에 맞는 문맥이 있다면 문맥을 가져가며 없다면 생성해서 2초동 안 고캐시에 저장합니다.

 

7. ResponseModel

package model

import "context"

type ResponseCtxModel struct {
	Value  string
	Ctx    context.Context
	Cancel context.CancelFunc
}

고캐시에 저장된 ResponseModel은 param에 하나씩 대응되는 context를 가지고 있어서 최초의 요청이 아닌 고루틴들을 관리합니다.

 

8. CacheRepositoryImpl

package repository

import (
	"fmt"
	"github.com/patrickmn/go-cache"
	"time"
)

type CacheRepositoryImpl struct {
	cache *cache.Cache
}

func (CacheRepositoryImpl) New() *CacheRepositoryImpl {
	c := cache.New(60*time.Second, 10*time.Second)
	return &CacheRepositoryImpl{c}
}

func (cacheRepositoryImpl *CacheRepositoryImpl) GetData(key string) string {
	data, found := cacheRepositoryImpl.cache.Get(key)
	if found {
		return data.(string)
	}
	return ""
}

func (cacheRepositoryImpl *CacheRepositoryImpl) SetData(key string, value string) {
	fmt.Printf("프록시서버: 본서버에서 response가 왔으므로 {%s: %s}의 데이터를 캐시합니다\n", key, value)
	cacheRepositoryImpl.cache.Set(key, value, cache.DefaultExpiration)

}

 

메인 캐시 저장소입니다. 본서버로부터 응답을 받으면 1분동안 데이터를 캐시합니다.