ChenKaiLiuG/GoEdgeNode
Archived
1
0
Fork 0
Code Issues Actions Packages Projects Releases Wiki Activity

实现缓存策略的部分功能

Browse Source
This commit is contained in:
GoEdgeLab
2020-10-05 16:55:14 +08:00
parent ae00133230
commit 870876bfff
12 changed files with 524 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/caches/file_writer.go
+77
View File
@@ -0,0 +1,77 @@
package caches
import (
"os"
"sync"
)
type FileWriter struct {
rawWriter *os.File
key string
size int64
expiredAt int64
locker *sync.RWMutex
isReleased bool
}
func NewFileWriter(rawWriter *os.File, key string, expiredAt int64, locker *sync.RWMutex) *FileWriter {
return &FileWriter{
key: key,
rawWriter: rawWriter,
expiredAt: expiredAt,
locker: locker,
}
}
// 写入数据
func (this *FileWriter) Write(data []byte) (n int, err error) {
n, err = this.rawWriter.Write(data)
this.size += int64(n)
if err != nil {
_ = this.rawWriter.Close()
_ = os.Remove(this.rawWriter.Name())
this.Release()
}
return
}
// 关闭
func (this *FileWriter) Close() error {
// 写入结束符
_, err := this.rawWriter.WriteString("\n$$$")
if err != nil {
_ = os.Remove(this.rawWriter.Name())
}
this.Release()
return err
}
// 丢弃
func (this *FileWriter) Discard() error {
err := os.Remove(this.rawWriter.Name())
this.Release()
return err
}
func (this *FileWriter) Size() int64 {
return this.size
}
func (this *FileWriter) ExpiredAt() int64 {
return this.expiredAt
}
func (this *FileWriter) Key() string {
return this.key
}
// 释放锁，一定要调用
func (this *FileWriter) Release() {
if this.isReleased {
return
}
this.isReleased = true
this.locker.Unlock()
}
internal/caches/manager.go
+2 -2
View File
@@ -123,9 +123,9 @@ func (this *Manager) FindStorageWithPolicy(policyId int64) StorageInterface {
// 根据策略获取存储对象
func (this *Manager) NewStorageWithPolicy(policy *serverconfigs.HTTPCachePolicy) StorageInterface {
switch policy.Type {
case serverconfigs.CachePolicyTypeFile:
case serverconfigs.CachePolicyStorageFile:
return NewFileStorage(policy)
case serverconfigs.CachePolicyTypeMemory:
case serverconfigs.CachePolicyStorageMemory:
return nil // TODO 暂时返回nil
}
return nil
internal/caches/storage_file.go
+9 -7
View File
@@ -34,7 +34,7 @@ var (
type FileStorage struct {
policy *serverconfigs.HTTPCachePolicy
cacheConfig *serverconfigs.HTTPFileCacheConfig
cacheConfig *serverconfigs.HTTPFileCacheStorage
list *List
locker sync.RWMutex
@@ -76,7 +76,7 @@ func (this *FileStorage) Init() error {
}()
// 配置
cacheConfig := &serverconfigs.HTTPFileCacheConfig{}
cacheConfig := &serverconfigs.HTTPFileCacheStorage{}
optionsJSON, err := json.Marshal(this.policy.Options)
if err != nil {
return err
@@ -119,7 +119,7 @@ func (this *FileStorage) Init() error {
return nil
}
func (this *FileStorage) Read(key string, readerBuf []byte, callback func(data []byte, expiredAt int64)) error {
func (this *FileStorage) Read(key string, readerBuf []byte, callback func(data []byte, size int64, expiredAt int64, isEOF bool)) error {
hash, path := this.keyPath(key)
if !this.list.Exist(hash) {
return ErrNotFound
@@ -185,6 +185,7 @@ func (this *FileStorage) Read(key string, readerBuf []byte, callback func(data [
}
startOffset := SizeExpiredAt + SizeKeyLength + keyLength + SizeNL
size := int(offset) + SizeEnd - startOffset
valueSize := offset - int64(startOffset)
_, err = fp.Seek(int64(startOffset), io.SeekStart)
if err != nil {
@@ -199,10 +200,10 @@ func (this *FileStorage) Read(key string, readerBuf []byte, callback func(data [
if n <= SizeEnd-size { // 已经到了末尾
break
} else {
callback(readerBuf[:n-(SizeEnd-size)], expiredAt)
callback(readerBuf[:n-(SizeEnd-size)], valueSize, expiredAt, true)
}
} else {
callback(readerBuf[:n], expiredAt)
callback(readerBuf[:n], valueSize, expiredAt, false)
}
}
if err != nil {
@@ -218,7 +219,7 @@ func (this *FileStorage) Read(key string, readerBuf []byte, callback func(data [
}
// 打开缓存文件等待写入
func (this *FileStorage) Open(key string, expiredAt int64) (*Writer, error) {
func (this *FileStorage) Open(key string, expiredAt int64) (Writer, error) {
hash := stringutil.Md5(key)
dir := this.cacheConfig.Dir + "/p" + strconv.FormatInt(this.policy.Id, 10) + "/" + hash[:2] + "/" + hash[2:4]
_, err := os.Stat(dir)
@@ -277,7 +278,7 @@ func (this *FileStorage) Open(key string, expiredAt int64) (*Writer, error) {
isOk = true
return NewWriter(writer, key, expiredAt, &this.locker), nil
return NewFileWriter(writer, key, expiredAt, &this.locker), nil
}
// 写入缓存数据
@@ -289,6 +290,7 @@ func (this *FileStorage) Write(key string, expiredAt int64, valueReader io.Reade
hash := stringutil.Md5(key)
dir := this.cacheConfig.Dir + "/p" + strconv.FormatInt(this.policy.Id, 10) + "/" + hash[:2] + "/" + hash[2:4]
_, err := os.Stat(dir)
if err != nil {
if !os.IsNotExist(err) {
internal/caches/storage_interface.go
+8 -3
View File
@@ -1,6 +1,8 @@
package caches
import "github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
import (
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
)
// 缓存存储接口
type StorageInterface interface {
@@ -8,10 +10,10 @@ type StorageInterface interface {
Init() error
// 读取缓存
Read(key string, readerBuf []byte, callback func(data []byte, expiredAt int64)) error
Read(key string, readerBuf []byte, callback func(data []byte, size int64, expiredAt int64, isEOF bool)) error
// 打开缓存写入器等待写入
Open(key string, expiredAt int64) (*Writer, error)
Open(key string, expiredAt int64) (Writer, error)
// 删除某个键值对应的缓存
Delete(key string) error
@@ -30,4 +32,7 @@ type StorageInterface interface {
// 获取当前存储的Policy
Policy() *serverconfigs.HTTPCachePolicy
// 将缓存添加到列表
AddToList(item *Item)
}
internal/caches/writer.go
+12 -64
View File
@@ -1,71 +1,19 @@
package caches
import (
"os"
"sync"
)
// 缓存内容写入接口
type Writer interface {
// 写入数据
Write(data []byte) (n int, err error)
type Writer struct {
rawWriter *os.File
key string
size int64
expiredAt int64
locker *sync.RWMutex
isReleased bool
}
// 关闭
Close() error
func NewWriter(rawWriter *os.File, key string, expiredAt int64, locker *sync.RWMutex) *Writer {
return &Writer{
key: key,
rawWriter: rawWriter,
expiredAt: expiredAt,
locker: locker,
}
}
// 丢弃
Discard() error
// 写入数据
func (this *Writer) Write(data []byte) error {
n, err := this.rawWriter.Write(data)
this.size += int64(n)
if err != nil {
_ = this.rawWriter.Close()
_ = os.Remove(this.rawWriter.Name())
this.Release()
}
// Key
Key() string
return err
}
// 关闭
func (this *Writer) Close() error {
// 写入结束符
_, err := this.rawWriter.WriteString("\n$$$")
if err != nil {
_ = os.Remove(this.rawWriter.Name())
}
this.Release()
return err
}
func (this *Writer) Size() int64 {
return this.size
}
func (this *Writer) ExpiredAt() int64 {
return this.expiredAt
}
func (this *Writer) Key() string {
return this.key
}
// 释放锁，一定要调用
func (this *Writer) Release() {
if this.isReleased {
return
}
this.isReleased = true
this.locker.Unlock()
// 过期时间
ExpiredAt() int64
}
internal/caches/writer_gzip.go
+47
View File
@@ -0,0 +1,47 @@
package caches
import "compress/gzip"
type gzipWriter struct {
rawWriter Writer
writer *gzip.Writer
key string
expiredAt int64
}
func NewGzipWriter(gw Writer, key string, expiredAt int64) Writer {
return &gzipWriter{
rawWriter: gw,
writer: gzip.NewWriter(gw),
key: key,
expiredAt: expiredAt,
}
}
func (this *gzipWriter) Write(data []byte) (n int, err error) {
return this.writer.Write(data)
}
func (this *gzipWriter) Close() error {
err := this.writer.Close()
if err != nil {
return err
}
return this.rawWriter.Close()
}
func (this *gzipWriter) Discard() error {
err := this.writer.Close()
if err != nil {
return err
}
return this.rawWriter.Discard()
}
func (this *gzipWriter) Key() string {
return this.key
}
func (this *gzipWriter) ExpiredAt() int64 {
return this.expiredAt
}